Spark SQL で C-Store テーブルのデータを読み取り・書き込み - AnalyticDB

AnalyticDB for MySQL クラスターでは、Spark SQL を使用して、XUANWU エンジンを使用する C-Store テーブルの読み取りと書き込みを、リアルタイムまたはバッチモードで行えます。

前提条件

AnalyticDB for MySQL Enterprise Edition、Basic Edition、または Data Lakehouse Edition のクラスターが作成されていること。
説明
- Data Lakehouse Edition クラスターには、0 ACU を超える予約済みストレージリソースが必要です。
- Enterprise Edition クラスターには、0 ACU を超える予約済みリソースが必要です。
- Basic Edition クラスターには、0 ACU を超える予約済みリソースが必要です。
AnalyticDB for MySQL Enterprise Edition、Basic Edition、または Data Lakehouse Edition クラスター用にジョブリソースグループが作成されていること。
AnalyticDB for MySQL Enterprise Edition、Basic Edition、または Data Lakehouse Edition クラスター用にデータベースアカウントが作成されていること。

ステップ 1: データ開発への移動

AnalyticDB for MySQL コンソールにログインします。コンソールの左上隅でリージョンを選択します。左側のナビゲーションウィンドウで クラスターリスト をクリックします。管理するクラスターを見つけて、クラスター ID をクリックします。
左側のナビゲーションウィンドウで、[ジョブ開発] > [SQL 開発] を選択します。
[SQLConsole] タブで、Spark エンジンとジョブリソースグループを選択します。

ステップ 2: データベースと C-Store テーブルの作成

説明

以下の SQL ステートメントは、バッチ処理モードまたはインタラクティブモードで実行できます。詳細については、「Spark SQL 実行モード」をご参照ください。

以下のステートメントを実行してデータベースを作成します。
```
CREATE DATABASE spark_create_adb_db_test;
```

以下のステートメントを実行して C-Store テーブルを作成します。テーブル作成構文の詳細については、「Spark SQL を使用した内部テーブルの作成」をご参照ください。

CREATE TABLE spark_create_adb_db_test.test_adb_tbl (
  id int,
  name string COMMENT 'test_name',
  age int
) using adb TBLPROPERTIES('primaryKey'='id,age',
                'distributeType'='HASH', 
                'distributeColumns'='id', 
                'partitionType'='value', 
                'partitionColumn'='age',   
                'partitionCount'='120',
                'storagePolicy'='COLD'   
               );

ステップ 3: C-Store テーブルの読み書き

説明

以下の SQL ステートメントは、バッチ処理モードまたはインタラクティブモードで実行できます。詳細については、「Spark SQL 実行モード」をご参照ください。
Spark SQL ステートメントを実行すると、システムは実行の成功または失敗を示すメッセージのみを返します。データは返されません。[Spark Jar 開発] ページの [アプリケーションリスト] タブにある [ログ] でテーブルデータを表示します。詳細については、「Spark アプリケーション情報の表示」をご参照ください。

バッチモード (INSERT OVERWRITE) の操作では、Spark エンジンとジョブリソースグループを選択するだけで済みます。リアルタイムモード (INSERT INTO) の操作では、SQL ステートメントを実行するために、SET コマンドを使用してインタラクティブなリソースグループを指定する必要があります。

バッチモード (INSERT OVERWRITE)

注意事項

バッチモードでデータを読み書きする際は、以下の点に注意してください。

Spark SQL は、C-Store パーティションテーブルからの読み取り (SELECT) と書き込み (INSERT) をサポートしていますが、非パーティション化テーブルはサポートしていません。パーティションテーブルの作成方法の詳細については、「CREATE TABLE」をご参照ください。
テーブルがパーティション化されているかどうかにかかわらず、Spark SQL を使用して C-Store テーブルからデータを更新 (UPDATE) または削除 (DELETE) することはできません。
ホットデータをクエリする場合、まず ELASTIC_ENABLE_HOT_PARTITION_HAS_HDD_REPLICA 設定項目を変更し、その後手動で BUILD ジョブをトリガーする必要があります。SQL クエリ文は、BUILD ジョブが完了した後にのみ実行できます。そうしないと、SQL ステートメントは失敗します。
1. XIHE エンジンを使用して、以下の SQL ステートメントを実行し、設定項目を変更します。
```
SET adb_config ELASTIC_ENABLE_HOT_PARTITION_HAS_HDD_REPLICA=true;
```
2. 手動で BUILD ジョブをトリガーします。
  - 特定のパーティションで BUILD ジョブを実行します。
    重要
    特定のパーティションで BUILD ジョブを実行できるのは、V3.1.6.0 以降のクラスターのみです。
    マイナーバージョンを表示および更新するには、AnalyticDB for MySQL コンソールの クラスター情報 ページにある 構成情報 セクションに移動します。
    BUILD TABLE <table_name> force partitions='partition1,partition2';
  - テーブル全体で BUILD ジョブを実行します。
    重要
    すべてのパーティションのインデックスを再構築するには時間がかかるため、この機能はデフォルトで無効になっています。特定のパーティションで BUILD ジョブを実行することをお勧めします。この機能を有効にする場合は、システムリスクを慎重に評価し、チケットを送信してください。
    BUILD TABLE <table_name> force = true;
3. BUILD ジョブをトリガーした後、以下のステートメントを実行してジョブのステータスを確認します。
```
SELECT table_name, schema_name, status FROM INFORMATION_SCHEMA.KEPLER_META_BUILD_TASK ORDER BY create_time DESC LIMIT 10;
```

C-Store テーブルへのデータ書き込み。
AnalyticDB for MySQL は、INSERT OVERWRITE 文を使用して、C-Store テーブルにデータをオフラインで書き込むことをサポートしています。C-Store テーブルにデータを書き込むには、以下のいずれかの方法を選択できます：
- 方法 1: 静的パーティションへの書き込み
```
INSERT OVERWRITE spark_create_adb_db_test.test_adb_tbl partition(age=10) VALUES (1, 'bom');
```
- 方法 2: 動的パーティションへの書き込み
```
INSERT OVERWRITE spark_create_adb_db_test.test_adb_tbl partition (age) VALUES (1, 'bom', 10);
```
C-Store テーブルからのデータ読み取り。
```
SELECT * FROM spark_create_adb_db_test.test_adb_tbl;
```

リアルタイムモード (INSERT INTO)

AnalyticDB for MySQL では、INSERT INTO ステートメントを使用して C-Store テーブルにリアルタイムモードでデータを書き込むことができます。JDBC メソッド、View メソッド、または Catalog メソッドを使用して、テーブルの読み書きが可能です。

JDBC メソッド

-- Database account, password, and internal endpoint of the cluster.
conf spark.adb.username=user;
conf spark.adb.password=password;
conf spark.adb.endpoint=amv-bp1a74zh****.ads.aliyuncs.com:3306;
-- Enable ENI-based access.
SET spark.adb.eni.enabled=true;
SET spark.adb.eni.vswitchId=vsw-bp12ldm83z4zu9k4d****;
SET spark.adb.eni.securityGroupId=sg-bp1cdm3acizrgq6x****;
-- Use the JDBC mode.
SET spark.adb.useJdbc = true;
-- Set an interactive resource group for SQL execution.
SET spark.adb.resourceGroup=user_default;
-- Write data.
INSERT INTO spark_create_adb_db_test.test_adb_tbl VALUES (1, 'adb', 20);
-- Read data.
SELECT * FROM spark_create_adb_db_test.test_adb_tbl;

以下の表にパラメーターを示します。

パラメーター	説明
spark.adb.username	クラスターのデータベースアカウント。
spark.adb.password	データベースアカウントのパスワード。
spark.adb.endpoint	クラスターの内部エンドポイントとポート。形式は `amv-bp1a74zh****.ads.aliyuncs.com:3306` です。
spark.adb.eni.enabled	ENI ベースのアクセスを有効にするかどうかを指定します。データにアクセスするには、`spark.adb.eni.enabled` パラメーターを true に設定します。
spark.adb.eni.vswitchId	クラスターが属する VSwitch の ID。
spark.adb.eni.securityGroupId	クラスターのセキュリティグループの ID。既存のセキュリティグループを選択するか、新しいセキュリティグループを作成できます。重要セキュリティグループは、Enterprise Edition、Basic Edition、または Data Lakehouse Edition クラスターと同じ VPC 内にある必要があります。
spark.adb.useJdbc	JDBC メソッドを使用してデータにアクセスするかどうかを指定します。有効な値は次のとおりです。 true: はい。 false (デフォルト値): いいえ。リアルタイムモードでデータを書き込むには、このパラメーターを true に設定します。
spark.adb.resourceGroup	SQL ステートメントの実行に使用されるインタラクティブなリソースグループ。説明 Enterprise Edition、Basic Edition、または Data Lakehouse Edition クラスターを作成すると、`user_default` という名前のインタラクティブなリソースグループが自動的に作成されます。このリソースグループを使用するか、新しいインタラクティブなリソースグループを作成できます。リソースグループの作成方法の詳細については、「リソースグループの作成と管理」をご参照ください。

View メソッド

-- Enable ENI-based access.
SET spark.adb.eni.enabled=true;
SET spark.adb.eni.vswitchId=vsw-bp1sxxsodv28ey5dl****;   
SET spark.adb.eni.securityGroupId=sg-bp19mr685pmg4ihc****;    
-- Create a view.
CREATE TEMPORARY VIEW table_tmp
USING org.apache.spark.sql.jdbc
OPTIONS (
  url 'jdbc:mysql://amv-bp1a74zh****.ads.aliyuncs.com:3306/spark_create_adb_db_test?useServerPrepStmts=false&rewriteBatchedStatements=true',     
  dbtable 'spark_create_adb_db_test.test_adb_tbl',    
  user 'user',       
  password 'password****'      
);
-- Write data.
INSERT INTO table_tmp VALUES (1, 'adb', 20);
-- Query data.
SELECT * FROM table_tmp;

以下の表にパラメーターを示します。

パラメーター	説明
spark.adb.eni.enabled	ENI ベースのアクセスを有効にするかどうかを指定します。データにアクセスするには、`spark.adb.eni.enabled` パラメーターを true に設定します。
spark.adb.eni.vswitchId	クラスターが属する VSwitch の ID。
spark.adb.eni.securityGroupId	クラスターのセキュリティグループの ID。既存のセキュリティグループを選択するか、新しいセキュリティグループを作成できます。重要セキュリティグループは、Enterprise Edition、Basic Edition、または Data Lakehouse Edition クラスターと同じ VPC 内にある必要があります。
table_tmp	ビューの名前。このトピックでは、`table_tmp` を例として使用します。
USING org.apache.spark.sql.jdbc	このパラメーターの値は `USING org.apache.spark.sql.jdbc` に固定されています。
url	クラスターの JDBC 接続文字列。形式: `jdbc:mysql://<endpoint:port>/<db_name>?useServerPrepStmts=false&rewriteBatchedStatements=true`。ここで、 `endpoint`: クラスターの内部エンドポイント。 port: クラスターのポート番号。値は 3306 に固定されています。 `db_name`: クラスター内のデータベースの名前。 useServerPrepStmts=false&rewriteBatchedStatements=true: これはバッチ書き込みに必要な構成です。書き込みパフォーマンスを向上させ、クラスターへの負荷を軽減します。例: `jdbc:mysql://amv-bp1a74zh****.ads.aliyuncs.com:3306/spark_create_adb_db_test?useServerPrepStmts=false&rewriteBatchedStatements=true`。
dbtable	クラスター内のテーブルの名前。形式は `db_name.table_name` です。このトピックでは、`spark_create_adb_db_test.test_adb_tbl` を例として使用します。
user	クラスターのデータベースアカウント。
password	データベースアカウントのパスワード。

Catalog メソッド

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://amv-bp1a74zh****.ads.aliyuncs.com:3306/?useServerPrepStmts=false&rewriteBatchedStatements=true>;
SET spark.sql.catalog.jdbc.user=<user>;
SET spark.sql.catalog.jdbc.password=<password>;
use jdbc;
SELECT * FROM spark_create_adb_db_test.test_adb_tbl;

以下の表にパラメーターを示します。

パラメーター	説明
spark.sql.catalog.jdbc	Spark SQL がデータソースを構成するために使用するメソッド。このパラメーターの値は `org.apache.spark.sql.execution.datasources.v2.jdbc.JDBCTableCatalog` に固定されています。説明 `spark.sql.catalog.catalog_name` パラメーターでは、`catalog_name` 部分はカスタマイズ可能です。このトピックでは、`jdbc` を例として使用します。
spark.sql.catalog.jdbc.url	クラスターの JDBC URL は、`jdbc:mysql://<endpoint:port>/?useServerPrepStmts=false&rewriteBatchedStatements=true` の形式です。ここで、 `endpoint`: クラスターの内部接続アドレス。 `port`: クラスターのポート番号。値は 3306 に固定されています。 `useServerPrepStmts=false&rewriteBatchedStatements=true`: クラスターへのバッチ書き込みに必要な設定であり、書き込みパフォーマンスを向上させ、クラスターへの負荷を軽減します。例: `jdbc:mysql://amv-bp1a74zh****.ads.aliyuncs.com:3306/?useServerPrepStmts=false&rewriteBatchedStatements=true`。
spark.sql.catalog.jdbc.user	クラスターのデータベースアカウント。
spark.sql.catalog.jdbc.password	データベースアカウントのパスワード。

Spark パラメーター

Spark SQL を使用して C-Store テーブルからデータを読み取る場合や、C-Store テーブルにデータを書き込む場合、以下のパラメーターを設定できます。

パラメーター	説明	デフォルト値
spark.adb.write.batchSize	1 回のバッチで書き込むデータレコード数です。値は正の整数である必要があります。説明このパラメーターは、リアルタイムモードでデータを書き込む場合にのみ適用されます。	600
spark.adb.write.arrow.maxMemoryBufferSize	書き込み操作における最大メモリバッファーサイズです。値は正の整数である必要があります。単位：MB。説明このパラメーターは、バッチモードでデータを書き込む場合にのみ適用されます。	1024
spark.adb.write.arrow.maxRecordSizePerBatch	1 回のバッチで書き込む最大レコード数です。値は正の整数である必要があります。説明このパラメーターは、バッチモードでデータを書き込む場合にのみ適用されます。	500