Spark Doris Connector

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

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

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

版本兼容

编译与安装

在源码目录下执行：

1. sh build.sh 2.3.4 2.11 ## spark 2.3.4, scala 2.11
2. sh build.sh 3.1.2 2.12 ## spark 3.1.2, scala 2.12

编译成功后，会在 output/ 目录下生成文件 doris-spark-2.3.4-2.11-1.0.0-SNAPSHOT.jar。将此文件复制到 Spark 的 ClassPath 中即可使用 Spark-Doris-Connector。例如，Local 模式运行的 Spark，将此文件放入 jars/ 文件夹下。Yarn集群模式运行的Spark，则将此文件放入预部署包中。

使用Maven管理

添加依赖

1. <dependency>
2. <groupId>org.apache.doris</groupId>
3. <artifactId>doris-spark-connector</artifactId>
4. <version>2.3.4-2.11-SNAPSHOT</version>
5. </dependency>

注意

2.3.4-2.11 可以根据spark和scala 版本替换成 3.1.2-2.12

使用示例

读取

SQL

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. );

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

11. dorisSparkRDD.collect()

写入

SQL

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. );

10. INSERT INTO spark_doris VALUES ("VALUE1","VALUE2",...);
11. # or
12. INSERT INTO spark_doris SELECT * 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. .save()

19. ## stream sink(StructuredStreaming)
20. val kafkaSource = spark.readStream
21. .option("kafka.bootstrap.servers", "$YOUR_KAFKA_SERVERS")
22. .option("startingOffsets", "latest")
23. .option("subscribe", "$YOUR_KAFKA_TOPICS")
24. .format("kafka")
25. .load()
26. kafkaSource.selectExpr("CAST(key AS STRING)", "CAST(value as STRING)")
27. .writeStream
28. .format("doris")
29. .option("checkpointLocation", "$YOUR_CHECKPOINT_LOCATION")
30. .option("doris.table.identifier", "$YOUR_DORIS_DATABASE_NAME.$YOUR_DORIS_TABLE_NAME")
31. .option("doris.fenodes", "$YOUR_DORIS_FE_HOSTNAME:$YOUR_DORIS_FE_RESFUL_PORT")
32. .option("user", "$YOUR_DORIS_USERNAME")
33. .option("password", "$YOUR_DORIS_PASSWORD")
34. //其它选项
35. //指定你要写入的字段
36. .option("doris.write.fields","$YOUR_FIELDS_TO_WRITE")
37. .start()
38. .awaitTermination()

配置

通用配置项

SQL 和 Dataframe 专有配置

RDD 专有配置

Doris 和 Spark 列类型映射关系

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