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

版本兼容

ConnectorSparkDorisJavaScala
2.3.4-2.11.xx2.x0.12+82.11
3.1.2-2.12.xx3.x0.12.+82.12
3.2.0-2.12.xx3.2.x0.12.+82.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 版本的示例，可供参考，这里

配置

通用配置项

KeyDefault ValueComment
doris.fenodes—Doris FE http 地址，支持多个地址，使用逗号分隔
doris.table.identifier—Doris 表名，如：db1.tbl1
doris.request.retries3向Doris发送请求的重试次数
doris.request.connect.timeout.ms30000向Doris发送请求的连接超时时间
doris.request.read.timeout.ms30000向Doris发送请求的读取超时时间
doris.request.query.timeout.s3600查询doris的超时时间，默认值为1小时，-1表示无超时限制
doris.request.tablet.sizeInteger.MAX_VALUE一个RDD Partition对应的Doris Tablet个数。
此数值设置越小，则会生成越多的Partition。从而提升Spark侧的并行度，但同时会对Doris造成更大的压力。
doris.batch.size1024一次从BE读取数据的最大行数。增大此数值可减少Spark与Doris之间建立连接的次数。
从而减轻网络延迟所带来的额外时间开销。
doris.exec.mem.limit2147483648单个查询的内存限制。默认为 2GB，单位为字节
doris.deserialize.arrow.asyncfalse是否支持异步转换Arrow格式到spark-doris-connector迭代所需的RowBatch
doris.deserialize.queue.size64异步转换Arrow格式的内部处理队列，当doris.deserialize.arrow.async为true时生效
doris.write.fields—指定写入Doris表的字段或者字段顺序，多列之间使用逗号分隔。
默认写入时要按照Doris表字段顺序写入全部字段。
sink.batch.size10000单次写BE的最大行数
sink.max-retries1写BE失败之后的重试次数
sink.properties.*—Stream Load 的导入参数。
例如: ‘sink.properties.column_separator’ = ‘, ‘
doris.sink.task.partition.size—Doris写入任务对应的 Partition 个数。Spark RDD 经过过滤等操作，最后写入的 Partition 数可能会比较大，但每个 Partition 对应的记录数比较少，导致写入频率增加和计算资源浪费。
此数值设置越小，可以降低 Doris 写入频率，减少 Doris 合并压力。该参数配合 doris.sink.task.use.repartition 使用。
doris.sink.task.use.repartitionfalse是否采用 repartition 方式控制 Doris写入 Partition数。默认值为 false，采用 coalesce 方式控制（注意: 如果在写入之前没有 Spark action 算子，可能会导致整个计算并行度降低）。
如果设置为 true，则采用 repartition 方式（注意: 可设置最后 Partition 数，但会额外增加 shuffle 开销）。
doris.sink.batch.interval.ms50每个批次sink的间隔时间，单位 ms。

SQL 和 Dataframe 专有配置

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

RDD 专有配置

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

Doris 和 Spark 列类型映射关系

Doris TypeSpark Type
NULL_TYPEDataTypes.NullType
BOOLEANDataTypes.BooleanType
TINYINTDataTypes.ByteType
SMALLINTDataTypes.ShortType
INTDataTypes.IntegerType
BIGINTDataTypes.LongType
FLOATDataTypes.FloatType
DOUBLEDataTypes.DoubleType
DATEDataTypes.StringType1
DATETIMEDataTypes.StringType1
BINARYDataTypes.BinaryType
DECIMALDecimalType
CHARDataTypes.StringType
LARGEINTDataTypes.StringType
VARCHARDataTypes.StringType
DECIMALV2DecimalType
TIMEDataTypes.DoubleType
HLLUnsupported datatype
  • 注：Connector中，将DATE和DATETIME映射为String。由于Doris底层存储引擎处理逻辑，直接使用时间类型时，覆盖的时间范围无法满足需求。所以使用 String 类型直接返回对应的时间可读文本。