EMR Serverless Spark で Paimon テーブルを使用する - E-MapReduce

前提条件

ワークスペースが作成済みであること。詳細については、「ワークスペースの作成」をご参照ください。

手順

ステップ 1：SQL セッションの作成

[セッション] ページに移動します。
1. EMR コンソールにログインします。
2. 左側のナビゲーションウィンドウで、[EMR Serverless] > [Spark] を選択します。
3. [Spark] ページで、対象のワークスペースの名前をクリックします。
4. EMR Serverless Spark ページで、左側のナビゲーションウィンドウの Sessions をクリックします。
SQL Sessions ページで、Create SQL Session をクリックします。

Create SQL Session ページで、Spark Configuration セクションのパラメーターを設定し、Create をクリックします。詳細については、「SQL セッションの管理」をご参照ください。

Spark はカタログを介して Paimon テーブルを読み書きします。シナリオに基づいてカタログタイプを選択します。詳細については、「データカタログの管理」をご参照ください。

データカタログ

データカタログを使用する場合、セッションパラメーターは不要です。Add Catalog ページの Catalogs をクリックし、Spark SQL 開発環境でデータカタログを選択します。

説明

EMR エンジンバージョン esr-4.3.0 以降、esr-3.3.0 以降、または esr-2.7.0 以降を推奨します。

カスタムカタログ

Data Lake Formation (DLF)

spark.sql.catalog.<catalogName>                                 org.apache.paimon.spark.SparkCatalog
spark.sql.catalog.<catalogName>.metastore                       rest
spark.sql.catalog.<catalogName>.uri                             http://cn-hangzhou-vpc.dlf.aliyuncs.com
spark.sql.catalog.<catalogName>.warehouse                       <catalog_name>
spark.sql.catalog.<catalogName>.token.provider                  dlf
spark.sql.catalog.<catalogName>.dlf.access-key-id               <access_key_id>
spark.sql.catalog.<catalogName>.dlf.access-key-secret           <access_key_secret>

パラメーター：

パラメーター	説明	値の例
`spark.sql.catalog.<catalogName>`	カタログの実装。	固定値：`org.apache.paimon.spark.SparkCatalog`
`spark.sql.catalog.<catalogName>.metastore`	メタデータのストレージ方法。DLF REST API を使用するには、`rest` に設定します。	固定値：`rest`
`spark.sql.catalog.<catalogName>.uri`	DLF URI。形式: `http://<endpoint>-vpc.dlf.aliyuncs.com`。	`http://cn-hangzhou-vpc.dlf.aliyuncs.com`
`spark.sql.catalog.<catalogName>.warehouse`	ウェアハウスパス。DLF の場合、これをカタログ名に設定します。	`<catalog_name>`
`spark.sql.catalog.<catalogName>.token.provider`	認証プロバイダーです。DLF の場合は `dlf` に設定します。	固定値：`dlf`
`spark.sql.catalog.<catalogName>.dlf.access-key-id`	Alibaba Cloud アカウントまたは RAM ユーザーの AccessKey ID。	`<access_key_id>`
`spark.sql.catalog.<catalogName>.dlf.access-key-secret`	Alibaba Cloud アカウントまたは RAM ユーザーの AccessKey Secret。	`<access_key_secret>`

DLF 1.0

メタデータは DLF 1.0 に保存されます。

spark.sql.catalog.<catalogName>                          org.apache.paimon.spark.SparkCatalog
spark.sql.catalog.<catalogName>.metastore                dlf
spark.sql.catalog.<catalogName>.dlf.catalog.id           <catalog_name>
spark.sql.catalog.<catalogName>.dlf.catalog.endpoint     dlf-vpc.cn-hangzhou.aliyuncs.com

パラメーター：

パラメーター	説明	値の例
`spark.sql.catalog.<catalogName>`	カタログの実装。	固定値：`org.apache.paimon.spark.SparkCatalog`
`spark.sql.catalog.<catalogName>.metastore`	メタデータの保存方法。DLF をメタストアとして使用する場合は `dlf` に設定します。	固定値：`dlf`
`spark.sql.catalog.<catalogName>.dlf.catalog.id`	DLF のカタログの名前。	`<catalog_name>`
`spark.sql.catalog.<catalogName>.dlf.catalog.endpoint`	DLF エンドポイント。リージョンに基づいて DLF エンドポイントを選択します。	`dlf-vpc.cn-hangzhou.aliyuncs.com`

Hive Metastore

メタデータは、指定された Hive Metastore に保存されます。

spark.sql.catalog.<catalogName>                 org.apache.paimon.spark.SparkCatalog
spark.sql.catalog.<catalogName>.metastore       hive
spark.sql.catalog.<catalogName>.uri             thrift://<yourHMSUri>:<port>

パラメーター：

パラメーター	説明	値の例
`spark.sql.catalog.<catalogName>`	カタログの実装。	固定値：`org.apache.paimon.spark.SparkCatalog`
`spark.sql.catalog.<catalogName>.metastore`	メタストアのタイプ。Hive メタストアを使用するには、`hive` に設定します。	固定値：`hive`
`spark.sql.catalog.<catalogName>.uri`	Hive Metastore の URI です。フォーマット: `thrift://<Hive メタストアの IP アドレス>:9083`。 `<Hive metastore の IP アドレス>` は、Hive Metastore サービスの内部 IP アドレスです。外部メタストアを使用するには、「外部 Hive Metastore サービスに接続する」をご参照ください。	`thrift://192.168..:9083`

ファイルシステム

メタデータはファイルシステムに保存されます。

spark.sql.catalog.<catalogName>                 org.apache.paimon.spark.SparkCatalog
spark.sql.catalog.<catalogName>.metastore       filesystem
spark.sql.catalog.<catalogName>.warehouse       oss://<yourBucketName>/warehouse

パラメーター：

パラメーター	説明	値の例
`spark.sql.catalog.<catalogName>`	カタログの実装。	固定値：`org.apache.paimon.spark.SparkCatalog`
`spark.sql.catalog.<catalogName>.metastore`	メタストアのタイプ。ファイルシステムをメタストアとして使用するには、`filesystem` に設定します。	固定値：`filesystem`
`spark.sql.catalog.<catalogName>.warehouse`	ウェアハウスのパスです。この例では、`<yourBucketName>` は OSS バケットの名前です。	`oss://my-bucket/warehouse`

DLF、DLF 1.0、Hive など、複数のカタログを同時に設定できます。例：

# DLF カタログの設定
spark.sql.catalog.paimon_dlf                                 org.apache.paimon.spark.SparkCatalog
spark.sql.catalog.paimon_dlf.metastore                       rest
spark.sql.catalog.paimon_dlf.uri                             http://cn-hangzhou-vpc.dlf.aliyuncs.com
spark.sql.catalog.paimon_dlf.warehouse                       <catalog_name>
spark.sql.catalog.paimon_dlf.token.provider                  dlf
spark.sql.catalog.paimon_dlf.dlf.access-key-id               <access_key_id>
spark.sql.catalog.paimon_dlf.dlf.access-key-secret           <access_key_secret>

# DLF 1.0 カタログの設定
spark.sql.catalog.paimon_dlf_legacy                                 org.apache.paimon.spark.SparkCatalog
spark.sql.catalog.paimon_dlf_legacy.metastore                       dlf
spark.sql.catalog.paimon_dlf_legacy.dlf.catalog.id                  <catalog_name>
spark.sql.catalog.paimon_dlf_legacy.dlf.catalog.endpoint            dlf-vpc.cn-hangzhou.aliyuncs.com


# hive1 カタログの設定
spark.sql.catalog.paimon_hive1                                 org.apache.paimon.spark.SparkCatalog
spark.sql.catalog.paimon_hive1.metastore                       hive
spark.sql.catalog.paimon_hive1.uri                             thrift://<yourHMSUri-1>:<port>

# hive2 カタログの設定
spark.sql.catalog.paimon_hive2                                 org.apache.paimon.spark.SparkCatalog
spark.sql.catalog.paimon_hive2.metastore                       hive
spark.sql.catalog.paimon_hive2.uri                             thrift://<yourHMSUri-2>:<port>

ステップ 2：Paimon テーブルの読み書き

SQL 開発ページに移動します。

EMR Serverless Spark ページで、左側メニューの Data Development をクリックします。
[開発] タブで、アイコンをクリックします。
[作成] ダイアログボックスで、users_task などの名前を入力し、タイプはデフォルトの [SparkSQL] のままにして、[OK] をクリックします。

新しい Spark SQL タブ (users_task) に次のコードをコピーします。

Paimon カタログ

-- データベースの作成
CREATE DATABASE IF NOT EXISTS paimon.ss_paimon_db;             

-- Paimon テーブルの作成
CREATE TABLE paimon.ss_paimon_db.paimon_tbl (id INT, name STRING) USING paimon;

-- Paimon テーブルへのデータ書き込み
INSERT INTO paimon.ss_paimon_db.paimon_tbl VALUES (1, "a"), (2, "b"), (3, "c");

-- Paimon テーブルからの書き込み結果のクエリ
SELECT * FROM paimon.ss_paimon_db.paimon_tbl ORDER BY id;

-- データベースの削除
DROP DATABASE paimon.ss_paimon_db CASCADE;

データベースのドロップダウンリストからデータベースを選択し、セッションのドロップダウンリストから作成した SQL セッションを選択します。
Run をクリックしてジョブを実行します。ジョブは次の出力を返します。

よくある質問

テーブルに対して DELETE、UPDATE、または MERGE ステートメントを実行するとエラーが発生するのはなぜですか。

現象: DELETE、UPDATE、または MERGE ステートメントを実行すると、次のようなエラーが返されます。

Caused by: org.apache.spark.sql.AnalysisException: Table does not support deletes/updates/merge: <tableName>.
    at org.apache.spark.sql.errors.QueryCompilationErrors$.tableDoesNotSupportError(QueryCompilationErrors.scala:1391)

原因：テーブルのストレージ形式が行レベルの更新をサポートしていないか、必要な Spark 設定がありません。
解決策：
1. テーブルのタイプを確認します。
  
  次のコマンドを実行して、テーブルが Paimon テーブルであるかどうかを確認します。
```
SHOW CREATE TABLE <tableName>;
```
  出力に USING PAIMON が含まれている場合、テーブルは Paimon テーブルです。出力に別のストレージ形式 (USING hive など) が表示される場合は、その形式が行レベルの更新をサポートしているか確認してください。
2. Spark の設定を確認します。
  
  テーブルが Paimon テーブルである場合は、Spark Configuration セクションを確認し、次の設定が存在することを確認します。
```
spark.sql.extensions org.apache.paimon.spark.extensions.PaimonSparkSessionExtensions
```
  この設定がない場合は、Spark Configuration に追加します。

E-MapReduce:Paimon の使用

前提条件

手順

ステップ 1：SQL セッションの作成

データカタログ

カスタムカタログ

Data Lake Formation (DLF)

DLF 1.0

Hive Metastore

ファイルシステム

ステップ 2：Paimon テーブルの読み書き

Paimon カタログ

よくある質問

関連ドキュメント