E-MapReduce:Delta Lake の使用

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

1. [セッション] ページに移動します。

   1. EMR コンソールにログインします。

   2. 左側のナビゲーションウィンドウで、[EMR Serverless] > [Spark] を選択します。

   3. [Spark] ページで、対象のワークスペースの名前をクリックします。

   4. EMR Serverless Spark ページで、左側のナビゲーションウィンドウの Sessions をクリックします。

2. SQL Session ページで、Connect to SQL Session をクリックします。

3. Connect to SQL Session ページで、Spark Configuration セクションに、次のプロパティを追加し、create をクリックします。詳細については、「SQL セッションの管理」をご参照ください。

   メタデータは、現在のワークスペースのデフォルトのカタログに保存されます。デフォルトのカタログを外部の Hive メタストアに変更する場合は、「外部 Hive メタストアサービスへの接続」をご参照ください。

   spark.sql.extensions             io.delta.sql.DeltaSparkSessionExtension
   spark.sql.catalog.spark_catalog  org.apache.spark.sql.delta.catalog.DeltaCatalog

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

1. [Development] ページに移動します。

   EMR Serverless Spark ページで、左側のナビゲーションペインで Development をクリックします。

2. [開発] タブで、 アイコンをクリックします。

3. [作成] ダイアログボックスで、users_task などの名前を入力し、タイプはデフォルトの [SparkSQL] のままにして、[OK] をクリックします。

4. 新しい SparkSQL タブ (users_task) に次のコードをコピーします。

   CREATE DATABASE IF NOT EXISTS ss_delta_db;
   CREATE TABLE ss_delta_db.delta_tbl (id INT, name STRING) USING delta;
   INSERT INTO ss_delta_db.delta_tbl VALUES (1, "a"), (2, "b");
   SELECT id, name FROM ss_delta_db.delta_tbl ORDER BY id;

5. ドロップダウンリストから、データベースと先ほど作成した SQL セッションを選択します。

6. 実行 をクリックします。

   クエリは、id と name の 2 つの列と、id=1, name='a' と id=2, name='b' の 2 つのレコードを含むテーブルを返します。

ステップ 3：データの更新

Delta Lake は、UPDATE、DELETE、MERGE INTO など、さまざまなデータ操作言語 (DML) 操作をサポートしています。次のコードに例を示します。

-- 更新操作
UPDATE ss_delta_db.delta_tbl SET name = "a_v2" WHERE id = 1;
-- 削除操作
DELETE FROM ss_delta_db.delta_tbl WHERE id = 2;
-- マージ操作
-- 一時テーブルを作成し、データを挿入します。
CREATE TABLE ss_delta_db.tmp_tbl(id INT, name STRING) USING delta;
INSERT INTO ss_delta_db.tmp_tbl VALUES (1, "a_v3"), (3, "c");
-- MERGE INTO 操作を実行します。
MERGE INTO ss_delta_db.delta_tbl AS target
USING ss_delta_db.tmp_tbl AS source
ON target.id = source.id
WHEN MATCHED THEN UPDATE SET *
WHEN NOT MATCHED THEN INSERT *;
-- 結果を検証します。
SELECT * FROM ss_delta_db.delta_tbl ORDER BY id;

クエリは、id と name の 2 つの列と、id=1, name='a_v3' と id=3, name='c' の 2 つのレコードを含むテーブルを返します。

ステップ 4：リソースのクリーンアップ

テストが完了したら、ストレージ料金の発生を防ぐためにリソースをクリーンアップします。次のコマンドを実行します：

DROP TABLE ss_delta_db.delta_tbl;
DROP TABLE ss_delta_db.tmp_tbl;
DROP DATABASE ss_delta_db;