ed5342a [Feature] Add S3 TVF batch sink support (#371) by wudi · 21 hours ago master
6f02e20 [Test] Add Doris regression test programs (#370) by wudi · 12 days ago
38d9b45 [Feature] Support TLS connections in Spark connector (#369) by wudi · 2 weeks ago
278f1ff [Fix] Adapt Arrow timestamp conversion for DATETIME and TIMESTAMPTZ (#366) by wudi · 4 weeks ago
ff27c83 [Feature] Support Spark 4.1 (#365) by YangJie · 6 weeks ago

Spark Connector for Apache Doris

Spark Doris Connector

More information about compilation and usage, please visit Spark Doris Connector

License

How to Build

You need to copy customer_env.sh.tpl to customer_env.sh before build and you need to configure it before build.

git clone git@github.com:apache/doris-spark-connector.git
cd doris-spark-connector/spark-doris-connector
./build.sh

QuickStart

download and compile Spark Doris Connector from https://github.com/apache/doris-spark-connector, we suggest compile Spark Doris Connector by Doris official image。

$ docker pull apache/doris:build-env-ldb-toolchain-latest

the result of compile jar is like：spark-doris-connector-3.1_2.12-1.0.0-SNAPSHOT.jar
download spark for https://spark.apache.org/downloads.html .if in china there have a good choice of tencent link https://mirrors.cloud.tencent.com/apache/spark/spark-3.1.2/

#download
wget https://mirrors.cloud.tencent.com/apache/spark/spark-3.1.2/spark-3.1.2-bin-hadoop3.2.tgz
#decompression
tar -xzvf spark-3.1.2-bin-hadoop3.2.tgz

config Spark environment

vim /etc/profile
export SPARK_HOME=/your_parh/spark-3.1.2-bin-hadoop3.2
export PATH=$PATH:$SPARK_HOME/bin
source /etc/profile

copy spark-doris-connector-3.1_2.12-1.0.0-SNAPSHOT.jar to spark jars directory。

cp /your_path/spark-doris-connector/target/spark-doris-connector-3.1_2.12-1.0.0-SNAPSHOT.jar  $SPARK_HOME/jars

created doris database and table。

create database mongo_doris;
use mongo_doris;
CREATE TABLE data_sync_test_simple
 (
         _id VARCHAR(32) DEFAULT '',
         id VARCHAR(32) DEFAULT '',
         user_name VARCHAR(32) DEFAULT '',
         member_list VARCHAR(32) DEFAULT ''
 )
 DUPLICATE KEY(_id)
 DISTRIBUTED BY HASH(_id) BUCKETS 10
 PROPERTIES("replication_num" = "1");
INSERT INTO data_sync_test_simple VALUES ('1','1','alex','123');

Input this coed in spark-shell.

import org.apache.doris.spark._
val dorisSparkRDD = sc.dorisRDD(
  tableIdentifier = Some("mongo_doris.data_sync_test"),
  cfg = Some(Map(
    "doris.fenodes" -> "127.0.0.1:8030",
    "doris.request.auth.user" -> "root",
    "doris.request.auth.password" -> ""
  ))
)
dorisSparkRDD.collect()

mongo_doris:doris database name
data_sync_test:doris table mame.
doris.fenodes:doris FE IP:http_port
doris.request.auth.user:doris user name.
doris.request.auth.password:doris password

if Spark is Cluster model,upload Jar to HDFS，add doris-spark-connector jar HDFS URL in spark.yarn.jars.

spark.yarn.jars=hdfs:///spark-jars/doris-spark-connector-3.1.2-2.12-1.0.0.jar

Link：https://github.com/apache/doris/discussions/9486

in pyspark,input this code in pyspark shell command.

dorisSparkDF = spark.read.format("doris")
.option("doris.table.identifier", "mongo_doris.data_sync_test")
.option("doris.fenodes", "127.0.0.1:8030")
.option("user", "root")
.option("password", "")
.load()
# show 5 lines data 
dorisSparkDF.show(5)

TLS

The connector supports one-way TLS for Doris HTTP APIs and Stream Load, MySQL/JDBC metadata queries, BE Thrift reads, and Arrow Flight SQL reads. TLS is disabled by default.

spark.read.format("doris")
  .option("doris.fenodes", "fe.example.com:8040")
  .option("doris.table.identifier", "database.table")
  .option("user", "root")
  .option("password", "")
  .option("doris.enable.tls", "true")
  .option("doris.tls.ca-certificate-path", "/etc/doris-tls/ca.pem")
  .load()

Option	Default	Description
`doris.enable.tls`	`false`	Enables TLS for all Doris protocols not listed in `doris.tls.excluded-protocols`.
`doris.tls.ca-certificate-path`	empty	Path to a PEM file containing one or more trusted X.509 CA certificates. An empty value uses the JVM or client-library default trust roots.
`doris.tls.skip-hostname-verification`	`false`	Disables hostname verification while retaining CA verification. This mode is not supported by Arrow Flight SQL.
`doris.tls.excluded-protocols`	empty	Comma-separated plaintext exceptions: `http`, `mysql`, `thrift`, and `arrowflight`.

The connector validates certificate chains and hostnames by default and does not modify JVM-global TLS settings. Client certificates and mutual TLS are not supported.

The CA file must be available at the same path on the Spark driver and every executor. For YARN, distribute it with --files /local/path/ca.pem#ca.pem and configure doris.tls.ca-certificate-path=ca.pem. For Kubernetes, mount the CA from a Secret or volume into both driver and executor pods and configure the mounted path.

type convertion for writing to doris using arrow

doris	spark
BOOLEAN	BooleanType
TINYINT	ByteType
SMALLINT	ShortType
INT	IntegerType
BIGINT	LongType
LARGEINT	StringType
FLOAT	FloatType
DOUBLE	DoubleType
DECIMAL(M,D)	DecimalType(M,D)
DATE	DateType
DATETIME	TimestampType
CHAR(L)	StringType
VARCHAR(L)	StringType
STRING	StringType
ARRAY	ARRAY
MAP	MAP
STRUCT	STRUCT

Report issues or submit pull request

If you find any bugs, feel free to file a GitHub issue or fix it by submitting a pull request.

Contact Us

Name	Scope
dev@doris.apache.org	Development-related discussions	Subscribe	Unsubscribe	Archives