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端完成数据过滤，减少数据传输量。

版本兼容

Connector Spark Doris Java Scala
2.3.4-2.11.xx 2.x 0.12+ 8 2.11
3.1.2-2.12.xx 3.x 0.12.+ 8 2.12
3.2.0-2.12.xx 3.2.x 0.12.+ 8 2.12

编译与安装

准备工作

1.修改custom_env.sh.tpl文件，重命名为custom_env.sh

2.指定thrift安装目录

1. ##源文件内容
2. #export THRIFT_BIN=
3. #export MVN_BIN=
4. #export JAVA_HOME=

6. ##修改如下,MacOS为例
7. export THRIFT_BIN=/opt/homebrew/Cellar/thrift@0.13.0/0.13.0/bin/thrift
8. #export MVN_BIN=
9. #export JAVA_HOME=

11. 安装 `thrift` 0.13.0 版本(注意：`Doris` 0.15 和最新的版本基于 `thrift` 0.13.0 构建, 之前的版本依然使用`thrift` 0.9.3 构建)
12. Windows:
13. 1.下载：`http://archive.apache.org/dist/thrift/0.13.0/thrift-0.13.0.exe`(下载目录自己指定)
14. 2.修改thrift-0.13.0.exe 为 thrift

16. MacOS:
17. 1. 下载：`brew install thrift@0.13.0`
18. 2. 默认下载地址：/opt/homebrew/Cellar/thrift@0.13.0/0.13.0/bin/thrift

21. 注：MacOS执行 `brew install thrift@0.13.0` 可能会报找不到版本的错误，解决方法如下，在终端执行：
22. 1. `brew tap-new $USER/local-tap`
23. 2. `brew extract --version='0.13.0' thrift $USER/local-tap`
24. 3. `brew install thrift@0.13.0`
25. 参考链接: `https://gist.github.com/tonydeng/02e571f273d6cce4230dc8d5f394493c`

27. Linux:
28. 1.下载源码包：`wget https://archive.apache.org/dist/thrift/0.13.0/thrift-0.13.0.tar.gz`
29. 2.安装依赖：`yum install -y autoconf automake libtool cmake ncurses-devel openssl-devel lzo-devel zlib-devel gcc gcc-c++`
30. 3.`tar zxvf thrift-0.13.0.tar.gz`
31. 4.`cd thrift-0.13.0`
32. 5.`./configure --without-tests`
33. 6.`make`
34. 7.`make install`
35. 安装完成后查看版本：thrift --version
36. 注：如果编译过Doris，则不需要安装thrift,可以直接使用 $DORIS_HOME/thirdparty/installed/bin/thrift

在源码目录下执行：

1. sh build.sh --spark 2.3.4 --scala 2.11 ## spark 2.3.4, scala 2.11
2. sh build.sh --spark 3.1.2 --scala 2.12 ## spark 3.1.2, scala 2.12
3. sh build.sh --spark 3.2.0 --scala 2.12 \
4. --mvn-args "-Dnetty.version=4.1.68.Final -Dfasterxml.jackson.version=2.12.3" ## spark 3.2.0, scala 2.12

注：如果你是从 tag 检出的源码，则可以直接执行 sh build.sh --tag，而无需指定 spark 和 scala 的版本。因为 tag 源码中的版本是固定的。

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

例如将 doris-spark-2.3.4-2.11-1.0.0-SNAPSHOT.jar 上传到 hdfs并在spark.yarn.jars参数上添加 hdfs上的Jar包路径

  1. 上传doris-spark-connector-3.1.2-2.12-1.0.0.jar 到hdfs。
1. hdfs dfs -mkdir /spark-jars/
2. hdfs dfs -put /your_local_path/doris-spark-connector-3.1.2-2.12-1.0.0.jar /spark-jars/
  1. 在集群中添加doris-spark-connector-3.1.2-2.12-1.0.0.jar 依赖。
1. spark.yarn.jars=hdfs:///spark-jars/doris-spark-connector-3.1.2-2.12-1.0.0.jar

使用Maven管理

1. <dependency>
2. <groupId>org.apache.doris</groupId>
3. <artifactId>spark-doris-connector-3.1_2.12</artifactId>
4. <!--artifactId>spark-doris-connector-2.3_2.11</artifactId-->
5. <version>1.0.1</version>
6. </dependency>

注意

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

使用示例

读取

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()

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 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()

java示例

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

配置

通用配置项

Key Default Value Comment
doris.fenodes — Doris FE http 地址，支持多个地址，使用逗号分隔
doris.table.identifier — Doris 表名，如：db1.tbl1
doris.request.retries 3 向Doris发送请求的重试次数
doris.request.connect.timeout.ms 30000 向Doris发送请求的连接超时时间
doris.request.read.timeout.ms 30000 向Doris发送请求的读取超时时间
doris.request.query.timeout.s 3600 查询doris的超时时间，默认值为1小时，-1表示无超时限制
doris.request.tablet.size Integer.MAX_VALUE 一个RDD Partition对应的Doris Tablet个数。 此数值设置越小，则会生成越多的Partition。从而提升Spark侧的并行度，但同时会对Doris造成更大的压力。
doris.batch.size 1024 一次从BE读取数据的最大行数。增大此数值可减少Spark与Doris之间建立连接的次数。 从而减轻网络延迟所带来的额外时间开销。
doris.exec.mem.limit 2147483648 单个查询的内存限制。默认为 2GB，单位为字节
doris.deserialize.arrow.async false 是否支持异步转换Arrow格式到spark-doris-connector迭代所需的RowBatch
doris.deserialize.queue.size 64 异步转换Arrow格式的内部处理队列，当doris.deserialize.arrow.async为true时生效
doris.write.fields — 指定写入Doris表的字段或者字段顺序，多列之间使用逗号分隔。 默认写入时要按照Doris表字段顺序写入全部字段。
sink.batch.size 10000 单次写BE的最大行数
sink.max-retries 1 写BE失败之后的重试次数

SQL 和 Dataframe 专有配置

Key Default Value Comment
user — 访问Doris的用户名
password — 访问Doris的密码
doris.filter.query.in.max.count 100 谓词下推中，in表达式value列表元素最大数量。超过此数量，则in表达式条件过滤在Spark侧处理。

RDD 专有配置

Key Default Value Comment
doris.request.auth.user — 访问Doris的用户名
doris.request.auth.password — 访问Doris的密码
doris.read.field — 读取Doris表的列名列表，多列之间使用逗号分隔
doris.filter.query — 过滤读取数据的表达式，此表达式透传给Doris。Doris使用此表达式完成源端数据过滤。

Doris 和 Spark 列类型映射关系

Doris Type Spark Type
NULL_TYPE DataTypes.NullType
BOOLEAN DataTypes.BooleanType
TINYINT DataTypes.ByteType
SMALLINT DataTypes.ShortType
INT DataTypes.IntegerType
BIGINT DataTypes.LongType
FLOAT DataTypes.FloatType
DOUBLE DataTypes.DoubleType
DATE DataTypes.StringType1
DATETIME DataTypes.StringType1
BINARY DataTypes.BinaryType
DECIMAL DecimalType
CHAR DataTypes.StringType
LARGEINT DataTypes.StringType
VARCHAR DataTypes.StringType
DECIMALV2 DecimalType
TIME DataTypes.DoubleType
HLL Unsupported datatype
  • 注：Connector中，将DATE和DATETIME映射为String。由于Doris底层存储引擎处理逻辑，直接使用时间类型时，覆盖的时间范围无法满足需求。所以使用 String 类型直接返回对应的时间可读文本。