Spark Doris Connector

Spark Doris Connector 可以支持通过 Spark 读取 Doris 中存储的数据，也支持通过Spark写入数据到Doris。

代码库地址：<https://github.com/apache/doris-spark-connector>

• 支持从Doris中读取数据
• 支持Spark DataFrame批量/流式 写入Doris
• 可以将Doris表映射为DataFrame或者RDD，推荐使用DataFrame。
• 支持在Doris端完成数据过滤，减少数据传输量。

版本兼容

编译与安装

准备工作

1. 修改custom_env.sh.tpl文件，重命名为custom_env.sh
2. 在源码目录下执行： sh build.sh 根据提示输入你需要的 Scala 与 Spark 版本进行编译。

编译成功后，会在 dist 目录生成目标jar包，如：spark-doris-connector-3.2_2.12-1.2.0-SNAPSHOT.jar。 将此文件复制到 Spark 的 ClassPath 中即可使用 Spark-Doris-Connector。

例如，Local 模式运行的 Spark，将此文件放入 jars/ 文件夹下。Yarn集群模式运行的Spark，则将此文件放入预部署包中。

例如将 spark-doris-connector-3.2_2.12-1.2.0-SNAPSHOT.jar 上传到 hdfs 并在 spark.yarn.jars 参数上添加 hdfs 上的 Jar 包路径

1. 上传 spark-doris-connector-3.2_2.12-1.2.0-SNAPSHOT.jar 到hdfs。

1. hdfs dfs -mkdir /spark-jars/
2. hdfs dfs -put /your_local_path/spark-doris-connector-3.2_2.12-1.2.0-SNAPSHOT.jar /spark-jars/

1. 在集群中添加 spark-doris-connector-3.2_2.12-1.2.0-SNAPSHOT.jar 依赖。

1. spark.yarn.jars=hdfs:///spark-jars/spark-doris-connector-3.2_2.12-1.2.0-SNAPSHOT.jar

使用Maven管理

1. <dependency>
2. <groupId>org.apache.doris</groupId>
3. <artifactId>spark-doris-connector-3.4_2.12</artifactId>
4. <version>1.3.0</version>
5. </dependency>

注意

请根据不同的 Spark 和 Scala 版本替换相应的 Connector 版本。

使用示例

读取

SQL

1. CREATE
2. TEMPORARY VIEW spark_doris
3. USING doris
4. OPTIONS(
5. "table.identifier"="$YOUR_DORIS_DATABASE_NAME.$YOUR_DORIS_TABLE_NAME",
6. "fenodes"="$YOUR_DORIS_FE_HOSTNAME:$YOUR_DORIS_FE_RESFUL_PORT",
7. "user"="$YOUR_DORIS_USERNAME",
8. "password"="$YOUR_DORIS_PASSWORD"
9. );

11. SELECT *
12. FROM spark_doris;

DataFrame

1. val dorisSparkDF = spark.read.format("doris")
2. .option("doris.table.identifier", "$YOUR_DORIS_DATABASE_NAME.$YOUR_DORIS_TABLE_NAME")
3. .option("doris.fenodes", "$YOUR_DORIS_FE_HOSTNAME:$YOUR_DORIS_FE_RESFUL_PORT")
4. .option("user", "$YOUR_DORIS_USERNAME")
5. .option("password", "$YOUR_DORIS_PASSWORD")
6. .load()

8. dorisSparkDF.show(5)

RDD

1. import org.apache.doris.spark._

3. val dorisSparkRDD = sc.dorisRDD(
4. tableIdentifier = Some("$YOUR_DORIS_DATABASE_NAME.$YOUR_DORIS_TABLE_NAME"),
5. cfg = Some(Map(
6. "doris.fenodes" -> "$YOUR_DORIS_FE_HOSTNAME:$YOUR_DORIS_FE_RESFUL_PORT",
7. "doris.request.auth.user" -> "$YOUR_DORIS_USERNAME",
8. "doris.request.auth.password" -> "$YOUR_DORIS_PASSWORD"
9. ))
10. )

12. dorisSparkRDD.collect()

pySpark

1. dorisSparkDF = spark.read.format("doris")
2. .option("doris.table.identifier", "$YOUR_DORIS_DATABASE_NAME.$YOUR_DORIS_TABLE_NAME")
3. .option("doris.fenodes", "$YOUR_DORIS_FE_HOSTNAME:$YOUR_DORIS_FE_RESFUL_PORT")
4. .option("user", "$YOUR_DORIS_USERNAME")
5. .option("password", "$YOUR_DORIS_PASSWORD")
6. .load()
7. // show 5 lines data
8. dorisSparkDF.show(5)

写入

SQL

1. CREATE
2. TEMPORARY VIEW spark_doris
3. USING doris
4. OPTIONS(
5. "table.identifier"="$YOUR_DORIS_DATABASE_NAME.$YOUR_DORIS_TABLE_NAME",
6. "fenodes"="$YOUR_DORIS_FE_HOSTNAME:$YOUR_DORIS_FE_RESFUL_PORT",
7. "user"="$YOUR_DORIS_USERNAME",
8. "password"="$YOUR_DORIS_PASSWORD"
9. );

11. INSERT INTO spark_doris
12. VALUES ("VALUE1", "VALUE2", ...);
13. # or
14. INSERT INTO spark_doris
15. SELECT *
16. FROM YOUR_TABLE
17. # or
18. INSERT OVERWRITE
19. SELECT *
20. FROM YOUR_TABLE

DataFrame(batch/stream)

1. ## batch sink
2. val mockDataDF = List(
3. (3, "440403001005", "21.cn"),
4. (1, "4404030013005", "22.cn"),
5. (33, null, "23.cn")
6. ).toDF("id", "mi_code", "mi_name")
7. mockDataDF.show(5)

9. mockDataDF.write.format("doris")
10. .option("doris.table.identifier", "$YOUR_DORIS_DATABASE_NAME.$YOUR_DORIS_TABLE_NAME")
11. .option("doris.fenodes", "$YOUR_DORIS_FE_HOSTNAME:$YOUR_DORIS_FE_RESFUL_PORT")
12. .option("user", "$YOUR_DORIS_USERNAME")
13. .option("password", "$YOUR_DORIS_PASSWORD")
14. //其它选项
15. //指定你要写入的字段
16. .option("doris.write.fields", "$YOUR_FIELDS_TO_WRITE")
17. // 支持设置 Overwrite 模式来覆盖数据
18. // .option("save_mode", SaveMode.Overwrite)
19. .save()

21. ## stream sink(StructuredStreaming)

23. ### 结果 DataFrame 和 doris 表相同的结构化数据, 配置方式和批量模式一致。
24. val sourceDf = spark.readStream.
25. .format("your_own_stream_source")
26. .load()

28. val resultDf = sourceDf.<transformations>

30. resultDf.writeStream
31. .format("doris")
32. .option("checkpointLocation", "$YOUR_CHECKPOINT_LOCATION")
33. .option("doris.table.identifier", "$YOUR_DORIS_DATABASE_NAME.$YOUR_DORIS_TABLE_NAME")
34. .option("doris.fenodes", "$YOUR_DORIS_FE_HOSTNAME:$YOUR_DORIS_FE_RESFUL_PORT")
35. .option("user", "$YOUR_DORIS_USERNAME")
36. .option("password", "$YOUR_DORIS_PASSWORD")
37. .start()
38. .awaitTermination()

40. ### 结果 DataFrame 中存在某一列的数据可以直接写入的，比如符合导入规范的 Kafka 消息中的 value 值

42. val kafkaSource = spark.readStream
43. .format("kafka")
44. .option("kafka.bootstrap.servers", "$YOUR_KAFKA_SERVERS")
45. .option("startingOffsets", "latest")
46. .option("subscribe", "$YOUR_KAFKA_TOPICS")
47. .load()
48. kafkaSource.selectExpr("CAST(value as STRING)")
49. .writeStream
50. .format("doris")
51. .option("checkpointLocation", "$YOUR_CHECKPOINT_LOCATION")
52. .option("doris.table.identifier", "$YOUR_DORIS_DATABASE_NAME.$YOUR_DORIS_TABLE_NAME")
53. .option("doris.fenodes", "$YOUR_DORIS_FE_HOSTNAME:$YOUR_DORIS_FE_RESFUL_PORT")
54. .option("user", "$YOUR_DORIS_USERNAME")
55. .option("password", "$YOUR_DORIS_PASSWORD")
56. // 设置该选项可以将 Kafka 消息中的 value 列不经过处理直接写入
57. .option("doris.sink.streaming.passthrough", "true")
58. .option("doris.sink.properties.format", "json")
59. // 其他选项
60. .start()
61. .awaitTermination()

Java示例

samples/doris-demo/spark-demo/ 下提供了 Java 版本的示例，可供参考，这里

配置

通用配置项

SQL 和 Dataframe 专有配置

Structured Streaming 专有配置

RDD 专有配置

提示

1. 在 Spark SQL 中，通过 insert into 方式写入数据时，如果 doris 的目标表中包含 BITMAP 或 HLL 类型的数据时，需要设置参数 doris.ignore-type 为对应类型, 并通过 doris.write.fields 对列进行映射转换，使用方式如下：

BITMAP

1. CREATE TEMPORARY VIEW spark_doris 2. USING doris 3. OPTIONS( 4. "table.identifier"="$YOUR_DORIS_DATABASE_NAME.$YOUR_DORIS_TABLE_NAME", 5. "fenodes"="$YOUR_DORIS_FE_HOSTNAME:$YOUR_DORIS_FE_RESFUL_PORT", 6. "user"="$YOUR_DORIS_USERNAME", 7. "password"="$YOUR_DORIS_PASSWORD" 8. "doris.ignore-type"="bitmap", 9. "doris.write.fields"="col1,col2,col3,bitmap_col2=to_bitmap(col2),bitmap_col3=bitmap_hash(col3)" 10. );

HLL

1. CREATE TEMPORARY VIEW spark_doris 2. USING doris 3. OPTIONS( 4. "table.identifier"="$YOUR_DORIS_DATABASE_NAME.$YOUR_DORIS_TABLE_NAME", 5. "fenodes"="$YOUR_DORIS_FE_HOSTNAME:$YOUR_DORIS_FE_RESFUL_PORT", 6. "user"="$YOUR_DORIS_USERNAME", 7. "password"="$YOUR_DORIS_PASSWORD" 8. "doris.ignore-type"="hll", 9. "doris.write.fields"="col1,hll_col1=hll_hash(col1)" 10. ); 2. 从 1.3.0 版本开始， doris.sink.max-retries 配置项的默认值为 0，即默认不进行重试。 当设置该参数大于 0 时，会进行批次级别的失败重试，会在 Spark Executor 内存中缓存 doris.sink.batch.size 所配置大小的数据，可能需要适当增大内存分配。 3. 从 1.3.0 版本开始，支持 overwrite 模式写入（只支持全表级别的数据覆盖），具体使用方式如下

DataFrame

1. resultDf.format("doris") 2. .option("doris.fenodes","$YOUR_DORIS_FE_HOSTNAME:$YOUR_DORIS_FE_RESFUL_PORT") 3. // your own options 4. .option("save_mode", SaveMode.Overwrite) 5. .save()

SQL

1. INSERT OVERWRITE your_target_table 2. SELECT * FROM your_source_table

Doris 和 Spark 列类型映射关系

• 注：Connector 中，将DATETIME映射为String。由于Doris底层存储引擎处理逻辑，直接使用时间类型时，覆盖的时间范围无法满足需求。所以使用 String 类型直接返回对应的时间可读文本。