StarRocks 作为高性能分析型数据库，支持通过 Spark 读取或写入数据，您可以使用 Spark Connector 连接 Spark 与 StarRocks 实现数据导入与导出。目前 EMR 提供的 Spark Connector 版本支持原生和旁路（Bypass）两种不同的模式读写 StarRocks 数据，区别如下表所示：

Bypass 模式相对于原生模式具备如下优势：

• 实现了 Spark StarRocks Catalog，无需事先在 Spark 中建外表，即可在 Spark 中直接访问 StarRocks 库表元数据信息。
• 对于 StarRocks 存算分离集群，支持通过 Spark 直接读写 StarRocks 存储在对象存储系统中的数据文件，避免占用 StarRocks 集群计算资源，影响在线查询。

Bypass 模式

Bypass 模式目前已在 EMR on ECS、EMR on VKE，以及 Serverless Spark 等产品形态中提供开箱即用。

配置参数

使用说明

EMR on ECS

Spark SQL 方式

启动 Spark SQL 客户端需要设置的参数示例如下：

spark-sql \
    --conf spark.sql.extensions="com.starrocks.connector.spark.StarRocksExtensions" \
    --conf spark.sql.catalog.starrocks="com.starrocks.connector.spark.catalog.StarRocksCatalog" \
    --conf spark.sql.catalog.starrocks.batch.size=8092 \
    --conf spark.sql.catalog.starrocks.writer.mode="BYPASS" \
    --conf spark.sql.catalog.starrocks.reader.mode="BYPASS" \
    --conf spark.sql.catalog.starrocks.fe.http.url="http://192.168.6.7:8030" \
    --conf spark.sql.catalog.starrocks.fe.jdbc.url="jdbc:mysql://192.168.6.7:9030" \
    --conf spark.sql.catalog.starrocks.user="root" \
    --conf spark.sql.catalog.starrocks.password="******" \
    --conf spark.sql.catalog.starrocks.fs.s3a.endpoint="https://tos-s3-cn-beijing.ivolces.com" \
    --conf spark.sql.catalog.starrocks.fs.s3a.endpoint.region="cn-beijing" \
    --conf spark.sql.catalog.starrocks.fs.s3a.connection.ssl.enabled=true \
    --conf spark.sql.catalog.starrocks.fs.s3a.path.style.access=false \
    --conf spark.sql.catalog.starrocks.fs.s3a.retry.limit=27 \
    --conf spark.sql.catalog.starrocks.fs.s3a.access.key="******" \
    --conf spark.sql.catalog.starrocks.fs.s3a.secret.key="******" \
    --jars /opt/emr/current/spark/starrocks-connector/starrocks-spark-connector-3.5_2.12-1.1.2-ve-bypass-2.jar

进入 Spark SQL 终端后，您可以切换到 StarRocks Catalog，然后执行查询语句：

USE starrocks.demo;
SELECT * FROM example_table;

pySpark 方式

准备 demo.py 程序示例：

import sys
import time
from operator import add

from pyspark.sql import SparkSession
from pyspark.sql.functions import *

if __name__ == "__main__":
    spark = SparkSession\
        .builder\
        .appName("Bypass Read")\
        .config("spark.shuffle.service.enabled", "true")\
        .config("spark.sql.extensions", "com.starrocks.connector.spark.StarRocksExtensions")\
        .config("spark.sql.catalog.starrocks", "com.starrocks.connector.spark.catalog.StarRocksCatalog")\
        .config("spark.sql.catalog.starrocks.fe.http.url", "http://xxx:8030")\
        .config("spark.sql.catalog.starrocks.fe.jdbc.url", "jdbc:mysql://xxx:9030")\
        .config("spark.sql.catalog.starrocks.user", "root")\
        .config("spark.sql.catalog.starrocks.password", "******")\
        .config("spark.sql.catalog.starrocks.writer.mode","BYPASS") \
        .config("spark.sql.catalog.starrocks.reader.mode","BYPASS") \
        .config("spark.sql.catalog.starrocks.fs.s3a.endpoint", "https://tos-s3-cn-beijing.ivolces.com")\
        .config("spark.sql.catalog.starrocks.fs.s3a.endpoint.region", "cn-beijing")\
        .config("spark.sql.catalog.starrocks.fs.s3a.connection.ssl.enabled", "true")\
        .config("spark.sql.catalog.starrocks.fs.s3a.path.style.access", "false")\
        .config("spark.sql.catalog.starrocks.fs.s3a.retry.limit", "27")\
        .config("spark.sql.catalog.starrocks.fs.s3a.access.key", "******")\
        .config("spark.sql.catalog.starrocks.fs.s3a.secret.key", "******")\
        .getOrCreate()
    spark.sql("use starrocks.demo").show()   
    spark.sql("show tables").show()
    spark.sql("select count(1) from tb_all_primitivetype_read_duplicate ").show()
    spark.sql("insert into tb_all_primitivetype_read_duplicate select * from tb_all_primitivetype_read_duplicate limit 10 ").show()
    spark.sql("select count(1) from tb_all_primitivetype_read_duplicate ").show()
    spark.stop()

• 以 Client 模式提交执行

spark-submit \
    --conf spark.app.name="bypass-test" \
    --jars /opt/emr/current/spark/starrocks-connector/starrocks-spark-connector-3.5_2.12-1.1.2-ve-bypass-2.jar \
    --conf "spark.yarn.jars=hdfs://master-1-1.emr-58048cfb501d90d6411d.cn-beijing.emr-volces.com:8020/user/application/label-pipeline/spark-jars-v2/*,hdfs://master-1-1.emr-58048cfb501d90d6411d.cn-beijing.emr-volces.com:8020/user/application/label-pipeline/app-jars/*" \
    demo.py

• 以 Cluster 模式提交执行

spark-submit \
    --deploy-mode cluster \
    --conf spark.app.name="bypass-test" \
    --conf "spark.yarn.jars=hdfs://master-1-1.emr-58048cfb501d90d6411d.cn-beijing.emr-volces.com:8020/user/application/label-pipeline/spark-jars-v2/*,hdfs://master-1-1.emr-58048cfb501d90d6411d.cn-beijing.emr-volces.com:8020/user/application/label-pipeline/app-jars/*" \
    demo.py

EMR on VKE

在掌握了如何向 EMR on VKE 集群提交 Spark 作业之后，您可以修改作业 yaml 文件，添加如下配置以实现通过 Spark 读写 StarRocks 数据文件：

spec:
    deps:
      jars:
        # starrocks-spark-connector jar 文件，VKE 集群镜像已内置
        - local:///opt/spark/starrocks-connector/starrocks-spark-connector-3.5_2.12-1.1.2-ve-bypass-2.jar
    sparkConf:
      spark.sql.extensions: "com.starrocks.connector.spark.StarRocksExtensions"
      spark.sql.catalog.starrocks: "com.starrocks.connector.spark.catalog.StarRocksCatalog"
      # 开启直读直写模式
      spark.sql.catalog.starrocks.writer.modeL: "BYPASS"
      spark.sql.catalog.starrocks.reader.mode: "BYPASS"
      # 对应 StarRocks 集群的连接配置，即 FE 各节点的内网 IP 地址
      spark.sql.catalog.starrocks.fe.http.url: "http://192.168.0.180:8030,192.168.0.181:8030,192.168.0.178:8030"
      spark.sql.catalog.starrocks.fe.jdbc.url: "jdbc:mysql://192.168.0.180:9030,192.168.0.181:9030,192.168.0.178:9030"
      # 对应 StarRocks 集群的认证配置
      spark.sql.catalog.starrocks.user: "root"
      spark.sql.catalog.starrocks.password: "******"
      # 对应 StarRocks 存算分离集群 TOS 连接和认证配置
      spark.sql.catalog.starrocks.fs.s3a.endpoint: "https://tos-s3-cn-beijing.ivolces.com"
      spark.sql.catalog.starrocks.fs.s3a.endpoint.region: "cn-beijing"
      spark.sql.catalog.starrocks.fs.s3a.connection.ssl.enabled: "true"
      spark.sql.catalog.starrocks.fs.s3a.path.style.access: "false"
      spark.sql.catalog.starrocks.fs.s3a.access.key: "******"
      spark.sql.catalog.starrocks.fs.s3a.secret.key: "******"
      # 【推荐】指定加载 so 文件路径，如果不指定则会尝试自解压，有 30s 左右的时间开销
      spark.driver.extraJavaOptions: "-Dcom.starrocks.format.jni.lib.path=/opt/starrocks/native"
      spark.executor.extraJavaOptions: "-Dcom.starrocks.format.jni.lib.path=/opt/starrocks/native"
      # 【可选】用于设置写入时攒批粒度
      spark.sql.catalog.starrocks.batch.size: "8092"

EMR Serverless Spark

参考 Serverless Spark 读写 StarRocks 操作手册。

原生模式

配置参数

使用说明

本小节以导入数据到 StarRocks 明细表 examples.tb_duplicate_key 为例，该表的建表语句如下：

CREATE TABLE IF NOT EXISTS tb_duplicate_key
(
    event_time BIGINT       NOT NULL COMMENT 'timestamp of event',
    event_type INT          NOT NULL COMMENT 'type of event',
    user_id    INT          NOT NULL COMMENT 'id of user',
    device     VARCHAR(128) NULL     COMMENT 'device'
) ENGINE = OLAP DUPLICATE KEY(event_time, event_type)
DISTRIBUTED BY HASH(user_id)
PROPERTIES (
    'replication_num' = '3'
);

Spark SQL 方式

您可以直接通过 Spark SQL 形式将数据写入 StarRocks 对应数据表中，步骤如下：

1. 进入 Spark SQL 交互终端，参考 Spark SQL Client 使用方式 进入 Spark SQL 交互终端，需要额外添加如下参数：

--jars /opt/emr/current/spark/starrocks-connector/starrocks-spark-connector-3.5_2.12-1.1.2-ve-bypass-2.jar

2. 通过 CREATE TABLE 创建一张 StarRocks tb_duplicate_key 表的外表，不要求同名：

CREATE TABLE IF NOT EXISTS tb_duplicate_key 
USING starrocks 
OPTIONS
(
    "starrocks.fe.http.url"="192.168.10.2:8030",
    "starrocks.fe.jdbc.url"="jdbc:mysql://192.168.10.2:9030",
    "starrocks.table.identifier"="examples.tb_duplicate_key",
    "starrocks.user"="system_query_user",
    "starrocks.password"="******"
);

3. 通过 INSERT INTO 操作将数据插入外表：

INSERT INTO tb_duplicate_key
VALUES (1703128450, 1, 1001, 'PHONE'),
       (1703128451, 0, 1002, 'PAD'),
       (1703128452, 1, 1003, 'TV');

正常情况下，您可以在 StarRocks 中查询到刚刚由 Spark 侧写入的数据。

Spark DataFrame 方式

本小节以 Batch 任务为例，演示将内存中构造的数据通过 Spark DataFrame 方式导入 StarRocks 的 tb_duplicate_key 表。Scala 示例代码如下：

val spark = SparkSession
  .builder()
  .appName("load_data_example")
  .getOrCreate()

import spark.implicits._

// 模拟数据
val data = Seq(
  (1703128450, 1, 1001L, "PHONE"),
  (1703128451, 0, 1002L, "PAD"),
  (1703128452, 1, 1003L, "TV"),
)

// 将数据写入 StarRocks
val df = data.toDF("event_time", "event_type", "user_id", "device")
df.write
  .format("starrocks")
  .option("starrocks.fe.http.url", "192.168.10.2:8030")
  .option("starrocks.fe.jdbc.url", "jdbc:mysql://192.168.10.2:9030")
  .option("starrocks.table.identifier", "examples.tb_duplicate_key")
  .option("starrocks.user", "system_query_user")
  .option("starrocks.password", "******")
  .mode("append")
  .save()

关于如何提交 Spark 任务可以参考 Spark 使用文档，需要额外添加如下参数：

--jars /opt/emr/current/spark/starrocks-connector/starrocks-spark-connector-3.5_2.12-1.1.2-ve-bypass-2.jar

更多信息

您可以访问 StarRocks 官方文档 了解关于使用 Spark Connector 向 StarRocks 导入数据的更多介绍，以及如何使用 Spark Connector 读取 StarRocks 中的数据。