Spark Doris Connector

Spark Doris Connector can support reading data stored in Doris and writing data to Doris through Spark.

Github: <https://github.com/apache/incubator-doris-spark-connector>

  • Support reading data from Doris.
  • Support Spark DataFrame batch/stream writing data to Doris
  • You can map the Doris table to DataFrame or RDD, it is recommended to use DataFrame.
  • Support the completion of data filtering on the Doris side to reduce the amount of data transmission.

Version Compatibility

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

Build and Install

Ready to work

1.Modify the custom_env.sh.tpl file and rename it to custom_env.sh

2.Specify the thrift installation directory

1. ##source file content
2. #export THRIFT_BIN=
3. #export MVN_BIN=
4. #export JAVA_HOME=

6. ##amend as below,MacOS as an example
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. Install `thrift` 0.13.0 (Note: `Doris` 0.15 and the latest builds are based on `thrift` 0.13.0, previous versions are still built with `thrift` 0.9.3)
12. Windows:
13. 1. Download: `http://archive.apache.org/dist/thrift/0.13.0/thrift-0.13.0.exe`
14. 2. Modify thrift-0.13.0.exe to thrift

16. MacOS:
17. 1. Download: `brew install thrift@0.13.0`
18. 2. default address: /opt/homebrew/Cellar/thrift@0.13.0/0.13.0/bin/thrift

20. Note: Executing `brew install thrift@0.13.0` on MacOS may report an error that the version cannot be found. The solution is as follows, execute it in the terminal:
21. 1. `brew tap-new $USER/local-tap`
22. 2. `brew extract --version='0.13.0' thrift $USER/local-tap`
23. 3. `brew install thrift@0.13.0`
24. Reference link: `https://gist.github.com/tonydeng/02e571f273d6cce4230dc8d5f394493c`

26. Linux:
27. 1.Download source package：`wget https://archive.apache.org/dist/thrift/0.13.0/thrift-0.13.0.tar.gz`
28. 2.Install dependencies：`yum install -y autoconf automake libtool cmake ncurses-devel openssl-devel lzo-devel zlib-devel gcc gcc-c++`
29. 3.`tar zxvf thrift-0.13.0.tar.gz`
30. 4.`cd thrift-0.13.0`
31. 5.`./configure --without-tests`
32. 6.`make`
33. 7.`make install`
34. Check the version after installation is complete：thrift --version
35. Note: If you have compiled Doris, you do not need to install thrift, you can directly use $DORIS_HOME/thirdparty/installed/bin/thrift

Execute following command in source dir

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

Note: If you check out the source code from tag, you can just run sh build.sh —tag without specifying the spark and scala versions. This is because the version in the tag source code is fixed.

After successful compilation, the file doris-spark-2.3.4-2.11-1.0.0-SNAPSHOT.jar will be generated in the output/ directory. Copy this file to ClassPath in Spark to use Spark-Doris-Connector. For example, Spark running in Local mode, put this file in the jars/ folder. Spark running in Yarn cluster mode, put this file in the pre-deployment package ,for example upload doris-spark-2.3.4-2.11-1.0.0-SNAPSHOT.jar to hdfs and add hdfs file path in spark.yarn.jars.

  1. Upload doris-spark-connector-3.1.2-2.12-1.0.0.jar Jar to 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. Add doris-spark-connector-3.1.2-2.12-1.0.0.jar depence in Cluster.
1. spark.yarn.jars=hdfs:///spark-jars/doris-spark-connector-3.1.2-2.12-1.0.0.jar

Using 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>

Notes

Please replace the Connector version according to the different Spark and Scala versions.

Example

Read

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)

Write

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. //other options
15. //specify the fields to write
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. //other options
35. //specify the fields to write
36. .option("doris.write.fields","$YOUR_FIELDS_TO_WRITE")
37. .start()
38. .awaitTermination()

Configuration

General

Key Default Value Comment
doris.fenodes — Doris FE http address, support multiple addresses, separated by commas
doris.table.identifier — Doris table identifier, eg, db1.tbl1
doris.request.retries 3 Number of retries to send requests to Doris
doris.request.connect.timeout.ms 30000 Connection timeout for sending requests to Doris
doris.request.read.timeout.ms 30000 Read timeout for sending request to Doris
doris.request.query.timeout.s 3600 Query the timeout time of doris, the default is 1 hour, -1 means no timeout limit
doris.request.tablet.size Integer.MAX_VALUE The number of Doris Tablets corresponding to an RDD Partition. The smaller this value is set, the more partitions will be generated. This will increase the parallelism on the Spark side, but at the same time will cause greater pressure on Doris.
doris.batch.size 1024 The maximum number of rows to read data from BE at one time. Increasing this value can reduce the number of connections between Spark and Doris. Thereby reducing the extra time overhead caused by network delay.
doris.exec.mem.limit 2147483648 Memory limit for a single query. The default is 2GB, in bytes.
doris.deserialize.arrow.async false Whether to support asynchronous conversion of Arrow format to RowBatch required for spark-doris-connector iteration
doris.deserialize.queue.size 64 Asynchronous conversion of the internal processing queue in Arrow format takes effect when doris.deserialize.arrow.async is true
doris.write.fields — Specifies the fields (or the order of the fields) to write to the Doris table, fileds separated by commas. By default, all fields are written in the order of Doris table fields.
sink.batch.size 10000 Maximum number of lines in a single write BE
sink.max-retries 1 Number of retries after writing BE failed
sink.properties.* — The stream load parameters. eg: sink.properties.column_separator’ = ‘,’
doris.sink.task.partition.size — The number of partitions corresponding to the Writing task. After filtering and other operations, the number of partitions written in Spark RDD may be large, but the number of records corresponding to each Partition is relatively small, resulting in increased writing frequency and waste of computing resources. The smaller this value is set, the less Doris write frequency and less Doris merge pressure. It is generally used with doris.sink.task.use.repartition.
doris.sink.task.use.repartition false Whether to use repartition mode to control the number of partitions written by Doris. The default value is false, and coalesce is used (note: if there is no Spark action before the write, the whole computation will be less parallel). If it is set to true, then repartition is used (note: you can set the final number of partitions at the cost of shuffle).
doris.sink.batch.interval.ms 50 The interval time of each batch sink, unit ms.

SQL & Dataframe Configuration

Key Default Value Comment
user — Doris username
password — Doris password
doris.filter.query.in.max.count 100 In the predicate pushdown, the maximum number of elements in the in expression value list. If this number is exceeded, the in-expression conditional filtering is processed on the Spark side.

RDD Configuration

Key Default Value Comment
doris.request.auth.user — Doris username
doris.request.auth.password — Doris password
doris.read.field — List of column names in the Doris table, separated by commas
doris.filter.query — Filter expression of the query, which is transparently transmitted to Doris. Doris uses this expression to complete source-side data filtering.

Doris & Spark Column Type Mapping

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
  • Note: In Connector, DATE and DATETIME are mapped to String. Due to the processing logic of the Doris underlying storage engine, when the time type is used directly, the time range covered cannot meet the demand. So use String type to directly return the corresponding time readable text.