ApsaraDB for SelectDB:Spark を使用したデータのインポート

オプション 1：Maven 依存関係の追加

pom.xml に以下の依存関係を追加します。他のバージョンについては、「Maven Repository」をご参照ください。

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

オプション 2：JAR パッケージのダウンロード

ご利用の Spark および Scala のバージョンに対応する JAR パッケージをダウンロードします。他のバージョンについては、「Maven Repository」をご参照ください。

以下の JAR パッケージは Java 8 でコンパイルされています。異なる Java バージョンが必要な場合は、テクニカルサポートまでお問い合わせください。バージョン文字列は左から順に、Spark バージョン、Scala バージョン、Spark Doris コネクタのバージョンを表します。

ダウンロード後、ご利用の Spark クラスターモードに応じて JAR をデプロイします。

• ローカルクラスターモード：JAR を Spark インストールディレクトリの jars ディレクトリに配置します。

• YARN クラスターモード：JAR を事前デプロイメントパッケージとしてアップロードします。

  1. HDFS への JAR のアップロード：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. クラスター構成への依存関係の追加： spark.yarn.jars=hdfs:///spark-jars/spark-doris-connector-3.2_2.12-1.3.2.jar

SparkSQL

Spark SQL シェルを起動し、一時ビューを使用してタスクを送信します。

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

Spark シェルを起動し、書き込みタスクを送信します。

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

接続パラメーター

読み取りパラメーター

書き込みパラメーター

環境のセットアップ

1. Spark の構成

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

spark-doris-connector-3.2_2.12-1.3.2.jar を SPARK_HOME/jars ディレクトリに配置します。

2. 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;

テストデータを生成するには、Data Management (DMS) を使用します。

3. SelectDB におけるターゲットテーブルの作成

MySQL プロトコルを介して SelectDB インスタンスに接続し、以下のコマンドを実行します。

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. パブリックエンドポイントの申請

SelectDB インスタンスのパブリックエンドポイントを申請します。詳細については、「パブリックエンドポイントの申請または解放」をご参照ください。

5. Spark ホストの IP をホワイトリストに追加

Spark 環境のパブリック IP アドレスを SelectDB IP アドレスホワイトリストに追加します。詳細については、「IP アドレスホワイトリストの設定」をご参照ください。

Spark SQL を使用したインポート

1. Spark SQL シェルを起動します。

   bin/spark-sql

2. インポートタスクを送信します。

   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. ジョブの完了後、ApsaraDB for SelectDB コンソールにログインして、インポートされたデータを確認します。

DataFrame を使用したインポート

1. Spark シェルを起動します。

   bin/spark-shell

2. インポートタスクを送信します。

   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. ジョブの完了後、ApsaraDB for SelectDB コンソールにログインして、インポートされたデータを確認します。