AnalyticDB:Spark SQL を使用した MySQL からのデータ読み取り

前提条件

開始する前に、以下を準備してください。

• AnalyticDB for MySQL Data Lakehouse Edition クラスター

• AnalyticDB for MySQL クラスターと ApsaraDB RDS for MySQL インスタンスが同じリージョンにあること

• クラスター用にジョブリソースグループが作成されていること。「リソースグループの作成」をご参照ください。

• AnalyticDB for MySQL クラスターのデータベースアカウント:

  • Alibaba Cloud アカウント: 特権アカウント。詳細については、「特権アカウントの作成」をご参照ください。

  • Resource Access Management (RAM) ユーザー：特権アカウントおよび標準アカウントの両方であり、標準アカウントは RAM ユーザーに関連付けられます。詳細については、「データベースアカウントの作成」および「データベースアカウントと RAM ユーザーの関連付けまたは関連付け解除」をご参照ください。

• ApsaraDB RDS for MySQL インスタンスが、インスタンスのポートでのトラフィックを許可するインバウンドおよびアウトバウンドルールを持つセキュリティグループに追加されていること。「ApsaraDB RDS for MySQL インスタンスのセキュリティグループを設定する」および「セキュリティグループルールの追加」をご参照ください。

ステップ 1: データ準備

ApsaraDB RDS for MySQL インスタンスにデータベースとテーブルを作成し、サンプル行を挿入します。

CREATE DATABASE `db`;

CREATE TABLE `db`.`test` (
  `id` int(11) DEFAULT NULL,
  `first_name` varchar(32) DEFAULT NULL,
  `laster_name` varchar(32) DEFAULT NULL,
  `age` int(11) DEFAULT NULL
) ENGINE=InnoDB DEFAULT CHARSET=utf8;

INSERT INTO test VALUES(1,'a','b',5);
INSERT INTO test VALUES(2,'c','d',6);
INSERT INTO test VALUES(3,'e','f',7);

ステップ 2: (オプション) CA 証明書と ApsaraDB RDS for MySQL ドライバーのアップロード

SSL 暗号化が不要な場合は、このステップをスキップしてください。直接「ステップ 3: Spark SQL ジョブの送信」に進みます。

1. ApsaraDB RDS for MySQL インスタンスで SSL 暗号化を有効にし、CA 証明書をダウンロードします。詳細については、「SSL 暗号化機能の設定」をご参照ください。

   重要

   内部接続またはパブリック接続を暗号化できます。この例では内部接続を使用します。CA 証明書は 1 年間有効です。有効期限が切れる前に新しい証明書を生成してください。期限切れの証明書では SSL 接続を確立できません。

2. https://dev.mysql.com/downloads/connector/j/ から、ご利用の ApsaraDB RDS for MySQL エンジンバージョンに一致する JDBC ドライバーをダウンロードします。この例では、mysql-connector-java-8.0.29.jar を使用します。

3. CA 証明書パッケージを抽出します。JKS ファイルとドライバー JAR を Object Storage Service (OSS) にアップロードします。「オブジェクトのアップロード」をご参照ください。

ステップ 3: Spark SQL ジョブの送信

1. 「AnalyticDB for MySQL コンソール」にログインします。左上隅でリージョンを選択します。左側のナビゲーションウィンドウで、[クラスター] をクリックします。[Data Lakehouse Edition] タブで、クラスター ID をクリックします。

2. 左側のナビゲーションウィンドウで、[ジョブ開発 ＞ SQL 開発] を選択します。

3. [SQLConsole] タブで、Spark エンジンとジョブリソースグループを選択します。

4. アクセスモードの SQL を入力して実行します。

set spark.adb.eni.enabled=true;
set spark.adb.eni.vswitchId=<vsw-bp1sxxsodv28ey5dl****>;
set spark.adb.eni.securityGroupId=<sg-bp19mr685pmg4ihc****>;

CREATE TEMPORARY VIEW table_tmp
USING org.apache.spark.sql.jdbc
OPTIONS (
  url '<jdbc:mysql://rm-bp1k87323a7ia****.mysql.rds.aliyuncs.com:3306>',
  dbtable '<db.test>',
  user '<user>',
  password '<password>'
);

SELECT * FROM table_tmp;

add jar oss://<bucketname>/mysql-connector-java-8.0.11.jar;
set spark.app.name=SSL_RDS_SQL;
set spark.adb.eni.enabled=true;
set spark.adb.eni.vswitchId=<vsw-bp1sxxsodv28ey5dl****>;
set spark.adb.eni.securityGroupId=<sg-bp19mr685pmg4ihc****>;
set spark.kubernetes.driverEnv.ADB_SPARK_DOWNLOAD_FILES=oss://<bucketname>/ApsaraDB-CA-Chain.jks;
set spark.executorEnv.ADB_SPARK_DOWNLOAD_FILES=oss://<bucketname>/ApsaraDB-CA-Chain.jks;

CREATE TEMPORARY VIEW table_view
USING org.apache.spark.sql.jdbc
OPTIONS (
  url '<jdbc:mysql://rm-bp1k87323a7ia****.mysql.rds.aliyuncs.com:3306/?useSSL=true&clientCertificateKeyStoreUrl=file:///tmp/<bucketname>/folder/ApsaraDB-CA-Chain.jks&clientCertificateKeyStorePassword=apsaradb&trustCertificateKeyStoreUrl=file:///tmp/<bucketname>/folder/ApsaraDB-CA-Chain.jks&trustCertificateKeyStorePassword=apsaradb&trustCertificateKeyStoreType=JKS&clientCertificateKeyStoreType=JKS>',
  dbtable '<db.test>',
  user '<user>',
  password '<password>'
);

set spark.adb.eni.enabled=true;
set spark.adb.eni.vswitchId=<vsw-bp1d14ddiw46fkgu1****>;
set spark.adb.eni.securityGroupId=<sg-bp19varsa8j0hyb****>;
set spark.sql.catalog.jdbc=org.apache.spark.sql.execution.datasources.v2.jdbc.JDBCTableCatalog;
set spark.sql.catalog.jdbc.url=<jdbc:mysql://rm-bp11mpql1e01l****.mysql.rds.aliyuncs.com:3306>;
set spark.sql.catalog.jdbc.user=<user>;
set spark.sql.catalog.jdbc.password=<password>;
USE jdbc;
SELECT * FROM db.test;