Import data using Spark

Option 1: Add the Maven dependency

Add the following dependency to your pom.xml. For other versions, see Maven Repository.

<dependency>
    <groupId>org.apache.doris</groupId>
    <artifactId>spark-doris-connector-3.2_2.12</artifactId>
    <version>1.3.2</version>
</dependency>

Option 2: Download a JAR package

Download the JAR package that matches your Spark and Scala version. For other versions, see Maven Repository.

The following JAR packages are compiled with Java 8. Contact technical support if you need a different Java version. The version string encodes, from left to right: Spark version, Scala version, and Spark Doris Connector version.

After downloading, deploy the JAR based on your Spark cluster mode:

• Local cluster mode: Place the JAR in the jars directory of your Spark installation.

• YARN cluster mode: Upload the JAR as a pre-deployment package.

  1. Upload the JAR to HDFS: ``bash hdfs dfs -mkdir /spark-jars/ hdfs dfs -put /<your_local_path>/spark-doris-connector-3.2_2.12-1.3.2.jar /spark-jars/ ``

  2. Add the dependency to your cluster configuration: `` spark.yarn.jars=hdfs:///spark-jars/spark-doris-connector-3.2_2.12-1.3.2.jar ``

Use Spark SQL

Start the Spark SQL shell and submit a task using a temporary view:

bin/spark-sql

val selectdbHttpPort = "selectdb-cn-****.selectdbfe.rds.aliyuncs.com:8080"
val selectdbJdbc = "jdbc:mysql://selectdb-cn-****.selectdbfe.rds.aliyuncs.com:9030"
val selectdbUser = "admin"
val selectdbPwd = "****"
val selectdbTable = "test_db.test_order"

CREATE TEMPORARY VIEW test_order
USING doris
OPTIONS(
 "table.identifier"="${selectdbTable}",
 "fenodes"="${selectdbHttpPort}",
 "user"="${selectdbUser}",
 "password"="${selectdbPwd}",
 "sink.properties.format"="json"
);

INSERT INTO test_order SELECT order_id, order_amount, order_status FROM tmp_tb;

Use DataFrame

Start the Spark shell and submit a write task:

bin/spark-shell

val spark = SparkSession.builder().master("local[1]").getOrCreate()
val df = spark.createDataFrame(Seq(
  ("1", 100, "Pending Payment"),
  ("2", 200, null),
  ("3", 300, "Received")
)).toDF("order_id", "order_amount", "order_status")

df.write
  .format("doris")
  .option("fenodes", selectdbHttpPort)
  .option("table.identifier", selectdbTable)
  .option("user", selectdbUser)
  .option("password", selectdbPwd)
  .option("sink.batch.size", 100000)
  .option("sink.max-retries", 3)
  .option("sink.properties.file.column_separator", "\t")
  .option("sink.properties.file.line_delimiter", "\n")
  .save()

Connection parameters

Read parameters

Write parameters

Set up the environment

1. Configure Spark.

Download and extract the Spark installation package:

wget https://archive.apache.org/dist/spark/spark-3.1.2/spark-3.1.2-bin-hadoop3.2.tgz
tar xvzf spark-3.1.2-bin-hadoop3.2.tgz

Place spark-doris-connector-3.2_2.12-1.3.2.jar in the SPARK_HOME/jars directory.

2. Create the source table in MySQL.

CREATE TABLE `employees` (
  `emp_no` int NOT NULL,
  `birth_date` date NOT NULL,
  `first_name` varchar(14) NOT NULL,
  `last_name` varchar(16) NOT NULL,
  `gender` enum('M','F') NOT NULL,
  `hire_date` date NOT NULL,
  PRIMARY KEY (`emp_no`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb3;

Use Data Management (DMS) to generate test data.

3. Create the target table in SelectDB.

Connect to your SelectDB instance over the MySQL protocol, then run:

CREATE DATABASE test_db;

USE test_db;
CREATE TABLE employees (
    emp_no       int NOT NULL,
    birth_date   date,
    first_name   varchar(20),
    last_name    varchar(20),
    gender       char(2),
    hire_date    date
)
UNIQUE KEY(`emp_no`)
DISTRIBUTED BY HASH(`emp_no`) BUCKETS 32;

4. Apply for a public endpoint.

Apply for a public endpoint for your SelectDB instance. See Apply for or release a public endpoint.

5. Add your Spark host IP to the whitelist.

Add the public IP address of your Spark environment to the SelectDB IP address whitelist. See Configure an IP address whitelist.

Import using Spark SQL

1. Start the Spark SQL shell:

   bin/spark-sql

2. Submit the import task:

   CREATE TEMPORARY VIEW mysql_tbl
   USING jdbc
   OPTIONS(
    "url"="jdbc:mysql://host:port/test_db",
    "dbtable"="employees",
    "driver"="com.mysql.jdbc.Driver",
    "user"="admin",
    "password"="****"
   );
   
   CREATE TEMPORARY VIEW selectdb_tbl
   USING doris
   OPTIONS(
    "table.identifier"="test_db.employees",
    "fenodes"="selectdb-cn-****-public.selectdbfe.rds.aliyuncs.com:8080",
    "user"="admin",
    "password"="****",
    "sink.properties.format"="json"
   );
   
   INSERT INTO selectdb_tbl SELECT emp_no, birth_date, first_name, last_name, gender, hire_date FROM mysql_tbl;

3. After the job completes, log on to the ApsaraDB for SelectDB console to verify the imported data.

Import using DataFrame

1. Start the Spark shell:

   bin/spark-shell

2. Submit the import task:

   val mysqlDF = spark.read.format("jdbc")
       .option("url", "jdbc:mysql://host:port/test_db")
       .option("dbtable", "employees")
       .option("driver", "com.mysql.jdbc.Driver")
       .option("user", "admin")
       .option("password", "****")
       .load()
   
   mysqlDF.write.format("doris")
       .option("fenodes", "host:httpPort")
       .option("table.identifier", "test_db.employees")
       .option("user", "admin")
       .option("password", "****")
       .option("sink.batch.size", 100000)
       .option("sink.max-retries", 3)
       .option("sink.properties.format", "json")
       .save()

3. After the job completes, log on to the ApsaraDB for SelectDB console to verify the imported data.