ApsaraDB for SelectDB:Impor\ data\ menggunakan\ Spark

Opsi 1: Tambahkan dependensi Maven

Tambahkan dependensi berikut ke file pom.xml Anda. Untuk versi lainnya, lihat Maven Repository.

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

Opsi 2: Unduh paket JAR

Unduh paket JAR yang sesuai dengan versi Spark dan Scala Anda. Untuk versi lainnya, lihat Maven Repository.

Paket JAR berikut dikompilasi dengan Java 8. Hubungi dukungan teknis jika Anda memerlukan versi Java yang berbeda. String versi menyertakan, dari kiri ke kanan: versi Spark, versi Scala, dan versi Spark Doris Connector.

Setelah mengunduh, deploy JAR berdasarkan mode kluster Spark Anda:

• Mode kluster lokal: Tempatkan file JAR di direktori jars instalasi Spark Anda.

• Mode kluster YARN: Unggah file JAR sebagai paket penyebaran awal.

  1. Unggah file JAR ke 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. Tambahkan dependensi ke konfigurasi kluster Anda: `` spark.yarn.jars=hdfs:///spark-jars/spark-doris-connector-3.2_2.12-1.3.2.jar ``

SparkSQL

Jalankan shell Spark SQL dan kirimkan tugas menggunakan tampilan temporary:

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;

DataFrame

Jalankan shell Spark dan kirimkan tugas penulisan:

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

Parameter koneksi

Parameter baca

Parameter tulis

Siapkan lingkungan

1. Konfigurasi Spark.

Unduh dan ekstrak paket instalasi Spark:

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

Tempatkan spark-doris-connector-3.2_2.12-1.3.2.jar di direktori SPARK_HOME/jars.

2. Buat tabel sumber di 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;

Gunakan Data Management (DMS) untuk menghasilkan data uji.

3. Buat tabel target di SelectDB.

Hubungkan ke instans SelectDB Anda melalui protokol MySQL, lalu jalankan:

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. Ajukan permohonan titik akhir publik.

Ajukan permohonan titik akhir publik untuk instans SelectDB Anda. Lihat Ajukan atau rilis titik akhir publik.

5. Tambahkan IP host Spark Anda ke daftar putih.

Tambahkan alamat IP publik lingkungan Spark Anda ke daftar putih alamat IP SelectDB. Lihat Konfigurasi daftar putih alamat IP.

Impor menggunakan Spark SQL

1. Jalankan shell Spark SQL:

   bin/spark-sql

2. Kirimkan tugas impor:

   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. Setelah pekerjaan selesai, login ke Konsol ApsaraDB for SelectDB untuk memverifikasi data yang diimpor.

Impor menggunakan DataFrame

1. Jalankan shell Spark:

   bin/spark-shell

2. Kirimkan tugas impor:

   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. Setelah pekerjaan selesai, login ke Konsol ApsaraDB for SelectDB untuk memverifikasi data yang diimpor.