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 toDataFrame or RDD, it is recommended to useDataFrame.
• Support the completion of data filtering on the Doris side to reduce the amount of data transmission.

Version Compatibility

Build and Install

Execute following command in source dir

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

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.

Using Maven

Add dependency

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>

Remark

2.3.4-2.11 can be repacled with 3.1.2-2.12 base on you spark and scala version

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

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

SQL & Dataframe Configuration

RDD Configuration

Doris & Spark Column Type Mapping

• Note: In Connector, DATE andDATETIME 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.