Iceberg の使用 - E-MapReduce - Alibaba Cloud ドキュメントセンター

Iceberg は、データレイク向けのオープンテーブルフォーマットです。Iceberg を使用すると、Hadoop 分散ファイルシステム (HDFS) や Alibaba Cloud Object Storage Service (OSS) 上にデータレイクストレージサービスを迅速に構築できます。このトピックでは、EMR Serverless Spark で Iceberg テーブルの読み書きを行う方法について説明します。

前提条件

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

手順

説明

SparkSQL とノートブックの両方で、Iceberg テーブルの読み書きがサポートされています。このトピックでは、SparkSQL ジョブを例として使用します。

ステップ 1：セッションリソースの作成

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

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

Spark はカタログを使用して Iceberg テーブルの読み書きを行います。ユースケースに基づいてカタログを選択できます。カタログの詳細については、「データカタログの管理」をご参照ください。

データカタログ

データカタログを使用する場合、セッションでパラメーターを設定する必要はありません。Catalog ページで、Add Catalog をクリックします。その後、SparkSQL 開発環境でデータカタログを直接選択できます。

説明

DLF (旧 DLF 2.5) の Iceberg にアクセスするには、エンジンバージョン esr-4.7.0、esr-3.6.0、またはそれ以降を使用します。
DLF 1.0 (旧 DLF-Legacy) または Hive メタストアの Iceberg にアクセスするには、エンジンバージョン esr-4.3.0、esr-3.3.0、esr-2.7.0、またはそれ以降を使用します。

カスタムカタログ

DLF (旧 DLF 2.5)

説明

エンジンバージョンの要件：esr-4.7.0、esr-3.6.0、またはそれ以降。

spark.sql.extensions org.apache.iceberg.spark.extensions.IcebergSparkSessionExtensions
spark.sql.catalog.iceberg_catalog org.apache.iceberg.spark.SparkCatalog
spark.sql.catalog.iceberg_catalog.catalog-impl org.apache.iceberg.rest.RESTCatalog
spark.sql.catalog.iceberg_catalog.uri http://<regionID>-vpc.dlf.aliyuncs.com
spark.sql.catalog.iceberg_catalog.warehouse  <catalog_name>
spark.sql.catalog.iceberg_catalog.io-impl org.apache.iceberg.rest.DlfFileIO
spark.sql.catalog.iceberg_catalog.rest.auth.type sigv4
spark.sql.catalog.iceberg_catalog.rest.auth.sigv4.delegate-auth-type none
spark.sql.catalog.iceberg_catalog.rest.signing-region <regionID>
spark.sql.catalog.iceberg_catalog.rest.signing-name DlfNext
spark.sql.catalog.iceberg_catalog.rest.access-key-id <access_key_id>
spark.sql.catalog.iceberg_catalog.rest.secret-access-key <access_key_secret>

以下の表でパラメーターについて説明します。

パラメーター	説明	値の例
`spark.sql.extensions`	Iceberg Spark 拡張機能を有効にします。	固定値： `org.apache.iceberg.spark.extensions.IcebergSparkSessionExtensions`
`spark.sql.catalog.iceberg_catalog`	`iceberg_catalog` という名前の Spark カタログを登録します。	固定値： `org.apache.iceberg.spark.SparkCatalog`
`spark.sql.catalog.iceberg_catalog.catalog-impl`	基盤となるカタログ実装として Iceberg REST カタログを指定します。	`org.apache.iceberg.rest.RESTCatalog`
`spark.sql.catalog.iceberg_catalog.uri`	DLF Iceberg サービスの REST API エンドポイント。フォーマットは `http://<regionID>-vpc.dlf.aliyuncs.com` です。	`http://cn-hangzhou-vpc.dlf.aliyuncs.com`
`spark.sql.catalog.iceberg_catalog.warehouse`	関連付けられた DLF カタログの名前。説明データ共有から作成された DLF カタログを関連付けないでください。	`<catalog_name>`
`spark.sql.catalog.iceberg_catalog.io-impl`	DLF 用にカスタマイズされた FileIO 実装を使用します。	固定値： `org.apache.iceberg.rest.DlfFileIO`
`spark.sql.catalog.iceberg_catalog.rest.auth.type`	REST リクエストを認証するための AWS 署名バージョン 4 (SigV4) 署名メカニズムを有効にします。	`sigv4`
`spark.sql.catalog.iceberg_catalog.rest.auth.sigv4.delegate-auth-type`	デリゲート認証を無効にします。クライアントはリクエストに署名するために AccessKey ID と AccessKey Secret を提供する必要があります。	`none`
`spark.sql.catalog.iceberg_catalog.rest.signing-region`	署名に使用されるリージョン。この値は、DLF サービスがデプロイされているリージョンと一致する必要があります。	`cn-hangzhou`
`spark.sql.catalog.iceberg_catalog.rest.signing-name`	署名に使用されるサービス名。	固定値： `DlfNext`
`spark.sql.catalog.iceberg_catalog.rest.access-key-id`	Alibaba Cloud アカウントまたは RAM ユーザーの AccessKey ID。	`<access_key_id>`
`spark.sql.catalog.iceberg_catalog.rest.secret-access-key`	Alibaba Cloud アカウントまたは RAM ユーザーの AccessKey Secret。	`<access_key_secret>`

DLF 1.0 (旧 DLF-Legacy)

説明

エンジンバージョンの要件：esr-4.3.0、esr-3.3.0、esr-2.7.0、またはそれ以降。

メタデータは DLF 1.0 (旧 DLF-Legacy) に保存されます。

spark.sql.extensions                           org.apache.iceberg.spark.extensions.IcebergSparkSessionExtensions
spark.sql.catalog.<catalogName>                org.apache.iceberg.spark.SparkCatalog
spark.sql.catalog.<catalogName>.catalog-impl   org.apache.iceberg.aliyun.dlf.hive.DlfCatalog
spark.sql.catalog.<catalogName>.dlf.catalog.id <catalog_name>

以下の表でパラメーターについて説明します。

パラメーター	説明	値の例
`spark.sql.extensions`	Iceberg Spark 拡張機能を有効にします。	固定値： `org.apache.iceberg.spark.extensions.IcebergSparkSessionExtensions`
`spark.sql.catalog.<catalogName>`	`<catalogName>` という名前のカタログを登録します。	固定値： `org.apache.iceberg.spark.SparkCatalog`
`spark.sql.catalog.<catalogName>.catalog-impl`	DLF 1.0 メタデータサービスに接続する、Alibaba Cloud DLF 1.0 Hive 互換実装を指定します。	固定値： `org.apache.iceberg.aliyun.dlf.hive.DlfCatalog`
`spark.sql.catalog.<catalogName>.dlf.catalog.id`	関連付けられた DLF カタログの名前を指定します。	`<catalog_name>`

Hive メタストア

メタデータは指定された Hive メタストアに保存されます。

spark.sql.extensions                           org.apache.iceberg.spark.extensions.IcebergSparkSessionExtensions
spark.sql.catalog.<catalogName>                org.apache.iceberg.spark.SparkCatalog
spark.sql.catalog.<catalogName>.catalog-impl   org.apache.iceberg.hive.HiveCatalog
spark.sql.catalog.<catalogName>.uri            thrift://<yourHMSUri>:<port>

以下の表でパラメーターについて説明します。

パラメーター	説明	値の例
`spark.sql.extensions`	Iceberg Spark 拡張機能を有効にします。	固定値： `org.apache.iceberg.spark.extensions.IcebergSparkSessionExtensions`
`spark.sql.catalog.<catalogName>`	`<catalogName>` という名前のカタログを登録します。	固定値： `org.apache.iceberg.spark.SparkCatalog`
`spark.sql.catalog.<catalogName>.catalog-impl`	公式の Iceberg `HiveCatalog` 実装を使用して、Hive メタストアのメタデータを管理するように指定します。	固定値： `org.apache.iceberg.hive.HiveCatalog`
`spark.sql.catalog.<catalogName>.uri`	Hive メタストアの URI。フォーマットは `thrift://<Hive metastore IP address>:9083` です。 `<Hive metastore IP address>` は HMS サービスの内部 IP アドレスです。外部のメタストアサービスを指定する必要がある場合は、「外部 Hive メタストアサービスへの接続」をご参照ください。	`thrift://192.168..:9083`

ファイルシステム

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

spark.sql.extensions                           org.apache.iceberg.spark.extensions.IcebergSparkSessionExtensions
spark.sql.catalog.<catalogName>                org.apache.iceberg.spark.SparkCatalog
spark.sql.catalog.<catalogName>.type           hadoop
spark.sql.catalog.<catalogName>.warehouse      oss://<yourBucketName>/warehouse

以下の表でパラメーターについて説明します。

パラメーター	説明	値の例
`spark.sql.extensions`	Iceberg Spark 拡張機能を有効にします。	固定値： `org.apache.iceberg.spark.extensions.IcebergSparkSessionExtensions`
`spark.sql.catalog.<catalogName>`	`<catalogName>` という名前のカタログを登録します。	固定値： `org.apache.iceberg.spark.SparkCatalog`
`spark.sql.catalog.<catalogName>.type`	カタログのタイプを `hadoop` として指定します。これは、HadoopCatalog がメタデータをファイルシステムに直接保存し、Hive Metastore を必要としないことを意味します。	`hadoop`
`spark.sql.catalog.<catalogName>.warehouse`	メタデータストレージのパスを指定します。このパスでは、`<yourBucketName>` は OSS バケットの名前を表します。	`oss://<yourBucketName>/warehouse`

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

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

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

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

説明

データベースを指定しない場合、テーブルはカタログの default データベースに作成されます。別のデータベースを作成して指定することもできます。

-- データベースを作成します。
CREATE DATABASE IF NOT EXISTS iceberg_catalog.db;
-- 非パーティションテーブルを作成します。
CREATE TABLE iceberg_catalog.db.tbl (
    id BIGINT NOT NULL COMMENT '一意の ID',
    data STRING
)
USING iceberg;
-- 非パーティションテーブルにデータを挿入します。
INSERT INTO iceberg_catalog.db.tbl VALUES
(1, 'Alice'),
(2, 'Bob'),
(3, 'Charlie');
-- 非パーティションテーブルからすべてのデータをクエリします。
SELECT * FROM iceberg_catalog.db.tbl;
-- 条件付きで非パーティションテーブルをクエリします。
SELECT * FROM iceberg_catalog.db.tbl WHERE id = 2;
-- 非パーティションテーブルのデータを更新します。
UPDATE iceberg_catalog.db.tbl SET data = 'David' WHERE id = 3;
-- 更新を確認するために再度クエリを実行します。
SELECT * FROM iceberg_catalog.db.tbl WHERE id = 3;
-- 非パーティションテーブルからデータを削除します。
DELETE FROM iceberg_catalog.db.tbl WHERE id = 1;
-- 削除を確認するために再度クエリを実行します。
SELECT * FROM iceberg_catalog.db.tbl;
-- パーティションテーブルを作成します。
CREATE TABLE iceberg_catalog.db.part_tbl (
    id BIGINT,
    data STRING,
    category STRING,
    ts TIMESTAMP,
    dt DATE
)
USING iceberg
PARTITIONED BY (dt, category);
-- パーティションテーブルにデータを挿入します。
INSERT INTO iceberg_catalog.db.part_tbl VALUES
  (1 , 'data-01', 'A', timestamp'2026-01-01 10:00:00', date'2026-01-01'),
  (2 , 'data-02', 'A', timestamp'2026-01-01 11:00:00', date'2026-01-01'),
  (3 , 'data-03', 'A', timestamp'2026-01-02 09:30:00', date'2026-01-02'),
  (4 , 'data-04', 'B', timestamp'2026-01-02 12:15:00', date'2026-01-02'),
  (5 , 'data-05', 'B', timestamp'2026-01-03 08:05:00', date'2026-01-03'),
  (6 , 'data-06', 'B', timestamp'2026-01-03 14:20:00', date'2026-01-03'),
  (7 , 'data-07', 'C', timestamp'2026-01-04 16:45:00', date'2026-01-04'),
  (8 , 'data-08', 'C', timestamp'2026-01-04 18:10:00', date'2026-01-04'),
  (9 , 'data-09', 'C', timestamp'2026-01-05 07:55:00', date'2026-01-05'),
  (10, 'data-10', 'A', timestamp'2026-01-05 13:35:00', date'2026-01-05');
-- パーティションテーブルからすべてのデータをクエリします。
SELECT * FROM iceberg_catalog.db.part_tbl;
-- dt='2026-01-01' のパーティションからデータをクエリします。
SELECT * FROM iceberg_catalog.db.part_tbl WHERE dt='2026-01-01';
-- 特定のカテゴリのデータをクエリします。
SELECT * FROM iceberg_catalog.db.part_tbl WHERE category = 'A';
-- 複数の条件 (日 + カテゴリ) でクエリを実行します。
SELECT * FROM iceberg_catalog.db.part_tbl 
WHERE dt='2026-01-01'
  AND category = 'A';
-- 各カテゴリのレコード数を集約してカウントします。
SELECT category, COUNT(*) AS count 
FROM iceberg_catalog.db.part_tbl 
GROUP BY category;
-- データベースをドロップします (注意して使用してください)。データベースをドロップする前に、テーブルが含まれていないことを確認してください。
-- DROP DATABASE iceberg_catalog.db;

セッションのドロップダウンリストで、作成した SQL セッションインスタンスを選択し、[Run] をクリックします。ジョブが完了すると、結果が下に表示されます。集約クエリを実行すると、出力は次のようになります：カテゴリ B (3)、カテゴリ C (3)、カテゴリ A (4)。

E-MapReduce:Iceberg の使用

前提条件

手順

ステップ 1：セッションリソースの作成

データカタログ

カスタムカタログ

DLF (旧 DLF 2.5)

DLF 1.0 (旧 DLF-Legacy)

Hive メタストア

ファイルシステム

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

関連ドキュメント