StarRocks Spark Connector

StarRocks 支持通过 Spark 读取或写入数据。您可以使用 Spark Connector 连接 Spark 与 StarRocks 实现数据导入，其原理是在内存中对数据进行攒批，按批次使用 Stream Load 将数据导入 StarRocks。Spark Connector 支持 DataFrame 和 SQL 接入形式，并支持 Batch 和 Structured Streaming 作业类型。

1 获取 Spark Connector

您可以从 Maven 中央仓库 中下载与您 Spark 版本匹配的最新的 spark-connector-starrocks.jar 文件，也可以使用由 EMR 团队提供的 Spark Connector 版本。

相对于开源版本的 Spark Connector，我们更加推荐您使用 EMR 团队提供的 Spark Connector 版本，相对而言优势如下：

1. 能够与 EMR 集群，及其周边生态更好的集成。

2. 增加一些 EMR 团队定制开发的竞争力特性。

3. 更加及时修复一些已知的 bug。

不过兼容开源是 EMR 作为开源大数据平台的基本原则，您仍然可以坚持使用开源 Spark Connector 版本。

2 使用方式

本小节以导入数据到 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'
);

2.1 Spark SQL 方式

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

1. 进入 Spark SQL 交互终端，参考 Spark SQL Client 使用方式 进入 Spark SQL 交互终端。

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

CREATE TABLE IF NOT EXISTS tb_duplicate_key 
USING starrocks 
OPTIONS
(
    "starrocks.fe.http.url"="{fe_ip}:8030",
    "starrocks.fe.jdbc.url"="jdbc:mysql://{fe_ip}: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 侧写入的数据。

2.2 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", "{fe_ip}:8030")
  .option("starrocks.fe.jdbc.url", "jdbc:mysql://{fe_ip}:9030")
  .option("starrocks.table.identifier", "examples.tb_duplicate_key")
  .option("starrocks.user", "system_query_user")
  .option("starrocks.password", "******")
  .mode("append")
  .save()

关于如何提交 Spark 任务可以参考 Spark 使用文档。

3 配置参数

4 更多信息

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