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

Build and Install

Ready to work

1. Modify the custom_env.sh.tpl file and rename it to custom_env.sh
2. Execute following command in source dir: sh build.sh Follow the prompts to enter the Scala and Spark versions you need to start compiling.

After the compilation is successful, the target jar package will be generated in the dist directory, such as: spark-doris-connector-3.1_2.12-1.2.0-SNAPSHOT.jar. 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 spark-doris-connector-3.1_2.12-1.2.0-SNAPSHOT.jar to hdfs and add hdfs file path in spark.yarn.jars.

1. Upload spark-doris-connector-3.1_2.12-1.2.0-SNAPSHOT.jar Jar to hdfs.

1. hdfs dfs -mkdir /spark-jars/
2. hdfs dfs -put /your_local_path/spark-doris-connector-3.1_2.12-1.2.0-SNAPSHOT.jar /spark-jars/

1. Add spark-doris-connector-3.1_2.12-1.2.0-SNAPSHOT.jar dependency in Cluster.

1. spark.yarn.jars=hdfs:///spark-jars/spark-doris-connector-3.1_2.12-1.2.0-SNAPSHOT.jar

Using Maven

1. <dependency>
2. <groupId>org.apache.doris</groupId>
3. <artifactId>spark-doris-connector-3.4_2.12</artifactId>
4. <version>1.3.0</version>
5. </dependency>

Notes

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

Example

Read

SQL

1. CREATE
2. TEMPORARY VIEW spark_doris
3. USING doris
4. OPTIONS(
5. "table.identifier"="$YOUR_DORIS_DATABASE_NAME.$YOUR_DORIS_TABLE_NAME",
6. "fenodes"="$YOUR_DORIS_FE_HOSTNAME:$YOUR_DORIS_FE_RESFUL_PORT",
7. "user"="$YOUR_DORIS_USERNAME",
8. "password"="$YOUR_DORIS_PASSWORD"
9. );

11. SELECT *
12. 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._

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

12. 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
2. TEMPORARY VIEW spark_doris
3. USING doris
4. OPTIONS(
5. "table.identifier"="$YOUR_DORIS_DATABASE_NAME.$YOUR_DORIS_TABLE_NAME",
6. "fenodes"="$YOUR_DORIS_FE_HOSTNAME:$YOUR_DORIS_FE_RESFUL_PORT",
7. "user"="$YOUR_DORIS_USERNAME",
8. "password"="$YOUR_DORIS_PASSWORD"
9. );

11. INSERT INTO spark_doris
12. VALUES ("VALUE1", "VALUE2", ...);
13. # or
14. INSERT INTO spark_doris
15. SELECT *
16. FROM YOUR_TABLE
17. # or
18. INSERT OVERWRITE
19. SELECT *
20. 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. // Support setting Overwrite mode to overwrite data
18. // .option("save_mode", SaveMode.Overwrite)
19. .save()

21. ## stream sink(StructuredStreaming)

23. ### Result DataFrame with structured data, the configuration method is the same as the batch mode.
24. val sourceDf = spark.readStream.
25. .format("your_own_stream_source")
26. .load()

28. val resultDf = sourceDf.<transformations>

30. resultDf.writeStream
31. .format("doris")
32. .option("checkpointLocation", "$YOUR_CHECKPOINT_LOCATION")
33. .option("doris.table.identifier", "$YOUR_DORIS_DATABASE_NAME.$YOUR_DORIS_TABLE_NAME")
34. .option("doris.fenodes", "$YOUR_DORIS_FE_HOSTNAME:$YOUR_DORIS_FE_RESFUL_PORT")
35. .option("user", "$YOUR_DORIS_USERNAME")
36. .option("password", "$YOUR_DORIS_PASSWORD")
37. .start()
38. .awaitTermination()

40. ### There is a column value in the Result DataFrame that can be written directly, such as the value in the kafka message that conforms to the import format

42. val kafkaSource = spark.readStream
43. .format("kafka")
44. .option("kafka.bootstrap.servers", "$YOUR_KAFKA_SERVERS")
45. .option("startingOffsets", "latest")
46. .option("subscribe", "$YOUR_KAFKA_TOPICS")
47. .load()
48. kafkaSource.selectExpr("CAST(key AS STRING)", "CAST(value as STRING)")
49. .writeStream
50. .format("doris")
51. .option("checkpointLocation", "$YOUR_CHECKPOINT_LOCATION")
52. .option("doris.table.identifier", "$YOUR_DORIS_DATABASE_NAME.$YOUR_DORIS_TABLE_NAME")
53. .option("doris.fenodes", "$YOUR_DORIS_FE_HOSTNAME:$YOUR_DORIS_FE_RESFUL_PORT")
54. .option("user", "$YOUR_DORIS_USERNAME")
55. .option("password", "$YOUR_DORIS_PASSWORD")
56. // Set this option to true, and the value column in the Kafka message will be written directly without processing.
57. .option("doris.sink.streaming.passthrough", "true")
58. .option("doris.sink.properties.format", "json")
59. //other options
60. .start()
61. .awaitTermination()

Configuration

General

SQL & Dataframe Configuration

Structured Streaming Configuration

RDD Configuration

tip

1. In Spark SQL, when writing data through insert into, if the target table of doris contains BITMAP or HLL type data, you need to set the parameter doris.ignore-type to the corresponding type, and set doris.write.fields maps the corresponding columns, the usage is as follows:

BITMAP

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. "doris.ignore-type"="bitmap", 9. "doris.write.fields"="col1,col2,col3,bitmap_col2=to_bitmap(col2),bitmap_col3=bitmap_hash(col3)" 10. );

HLL

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. "doris.ignore-type"="hll", 9. "doris.write.fields"="col1,hll_col1=hll_hash(col1)" 10. ); 2. Since version 1.3.0, the default value of doris.sink.max-retries configuration is 0, which means no retries are performed by default. When this parameter is set greater than 0, batch-level failure retries will be performed, and data of the configured size of doris.sink.batch.size will be cached in the Spark Executor memory. The memory allocation may need to be appropriately increased. 3. Since version 1.3.0, overwrite mode insertion is supported (only full table-level overwrite insertion is supported). The specific usage is as follows

DataFrame

1. resultDf.format("doris") 2. .option("doris.fenodes","$YOUR_DORIS_FE_HOSTNAME:$YOUR_DORIS_FE_RESFUL_PORT") 3. // your own options 4. .option("save_mode", SaveMode.Overwrite) 5. .save()

SQL

1. INSERT OVERWRITE your_target_table 2. SELECT * FROM your_source_table

Doris & Spark Column Type Mapping

• Note: In Connector, DATETIME is 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.