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

E-MapReduce では、Flink SQL を使用して Paimon からのデータ読み取りおよび Paimon へのデータ書き込みがサポートされています。本トピックでは、カタログの作成、ストリーミングによる読み書き、OLAP クエリの実行方法について例を示します。

前提条件

Flink および Paimon を選択した Dataflow クラスターまたはカスタムクラスターを作成済みである必要があります。詳細については、「クラスターの作成」をご参照ください。

説明

Hive カタログを使用する場合は、Flink、Paimon、Hive を選択したカスタムクラスターを作成する必要があります。また、Metadata Storage Method を Self-managed RDS または Built-in MySQL に設定する必要があります。

制限事項

E-MapReduce V3.46.0 および V5.17.0 は、DLF カタログおよび Hive カタログをサポートしていません。
E-MapReduce V3.46.0 ～ V3.50.X および E-MapReduce V5.12.0 ～ V5.16.X を実行するクラスターでは、Flink SQL を使用して Paimon の読み書きが可能です。

説明
E-MapReduce V3.51.X 以降および E-MapReduce V5.17.X 以降については、 Apache Paimon ドキュメントを参照し、ご利用の EMR クラスターで統合を設定してください。

操作手順

ステップ 1：依存関係の設定

ファイルシステムカタログ、Hive カタログ、または DLF カタログを使用して Paimon の読み書きが可能です。選択した方法に応じて依存関係を設定してください。

ファイルシステムカタログ

cp /opt/apps/PAIMON/paimon-current/lib/flink/*.jar /opt/apps/FLINK/flink-current/lib/

Hive カタログ

cp /opt/apps/PAIMON/paimon-current/lib/flink/*.jar /opt/apps/FLINK/flink-current/lib/
cp /opt/apps/FLINK/flink-current/opt/catalogs/hive-2.3.6/*.jar /opt/apps/FLINK/flink-current/lib/

DLF カタログ

cp /opt/apps/PAIMON/paimon-current/lib/flink/*.jar /opt/apps/FLINK/flink-current/lib/
cp /opt/apps/PAIMON/paimon-current/lib/jackson/*.jar /opt/apps/FLINK/flink-current/lib/
cp /opt/apps/METASTORE/metastore-*/hive2/*.jar /opt/apps/FLINK/flink-current/lib/
cp /opt/apps/FLINK/flink-current/opt/catalogs/hive-2.3.6/*.jar /opt/apps/FLINK/flink-current/lib/

ステップ 2：クラスターの起動

本トピックではセッションモードを例として説明します。その他のモードについては、「基本的な使用方法」をご参照ください。

次のコマンドを実行して、デタッチされた YARN セッションを起動します。

yarn-session.sh --detached

ステップ 3：カタログの作成

Paimon は、HDFS などのファイルシステムまたは OSS-HDFS などのオブジェクトストレージにデータおよびメタデータを保存します。warehouse パラメーターはルートパスを指定します。指定されたウェアハウスパスが存在しない場合、Paimon は自動的に作成します。パスがすでに存在する場合は、そのパス内の既存テーブルにカタログからアクセスできます。

また、メタデータを Hive または DLF に同期することで、他のサービスが Paimon データにアクセスできるようにすることも可能です。

説明

E-MapReduce V3.46.0 および V5.17.0 は、DLF カタログおよび Hive カタログをサポートしていません。

ファイルシステムカタログ

ファイルシステムカタログは、メタデータをファイルシステムまたはオブジェクトストレージ内にのみ保存します。

次のコマンドを実行して、Flink SQL クライアントを起動します。
```
sql-client.sh
```

次の Flink SQL ステートメントを実行して、ファイルシステムカタログを作成します。

CREATE CATALOG test_catalog WITH (
    'type' = 'paimon',
    'metastore' = 'filesystem',
    'warehouse' = 'oss://<yourBucketName>/warehouse'
);

Hive カタログ

Hive カタログは、メタデータを Hive Metastore に同期します。Hive カタログ内で作成されたテーブルは、Hive から直接クエリできます。

Hive から Paimon をクエリする方法の詳細については、「Paimon と Hive の統合」をご参照ください。

次のコマンドを実行して、Flink SQL クライアントを起動します。
```
sql-client.sh
```
説明
Hive のバージョンに関係なく、起動コマンドは同じです。

次の Flink SQL ステートメントを実行して、Hive カタログを作成します。

CREATE CATALOG test_catalog WITH (
    'type' = 'paimon',
    'metastore' = 'hive',
    'uri' = 'thrift://master-1-1:9083', -- uri パラメーターは Hive メタストアサービスのアドレスを指定します。
    'warehouse' = 'oss://<yourBucketName>/warehouse'
);

DLF カタログ

DLF カタログは、メタデータを DLF に同期します。

重要

クラスター作成時に、Metadata を DLF 統合メタデータ に設定する必要があります。

次のコマンドを実行して、Flink SQL クライアントを起動します。
```
sql-client.sh
```
説明
Hive のバージョンに関係なく、起動コマンドは同じです。

次の Flink SQL ステートメントを実行して、DLF カタログを作成します。

CREATE CATALOG test_catalog WITH (
    'type' = 'paimon',
    'metastore' = 'dlf',
    'hive-conf-dir' = '/etc/taihao-apps/flink-conf',
    'warehouse' = 'oss://<yourBucketName>/warehouse'
);

ステップ 4：ストリーミングによる Paimon の読み書き

次の Flink SQL ステートメントを実行して、カタログ内にテーブルを作成し、その後読み書きを行います。

-- 実行モードをストリーミングに設定します。
SET 'execution.runtime-mode' = 'streaming';

-- Paimon では、ストリーミングジョブに対してチェックポイント間隔を設定する必要があります。
SET 'execution.checkpointing.interval' = '10s';

-- 前のステップで作成したカタログを使用します。
USE CATALOG test_catalog;

-- テスト用データベースを作成して使用します。
CREATE DATABASE test_db;
USE test_db;

-- datagen を使用してランダムデータを生成します。
CREATE TEMPORARY TABLE datagen_source (
    uuid int,
    kind int,
    price int
) WITH (
    'connector' = 'datagen',
    'fields.kind.min' = '0',
    'fields.kind.max' = '9',
    'rows-per-second' = '10'
);

-- Paimon テーブルを作成します。
CREATE TABLE test_tbl (
    uuid int,
    kind int,
    price int,
    PRIMARY KEY (uuid) NOT ENFORCED
);

-- Paimon テーブルにデータを書き込みます。
INSERT INTO test_tbl SELECT * FROM datagen_source;

-- テーブルからデータを読み取ります。
-- 上記のストリーミング書き込みジョブは同時実行されます。
-- Flink クラスターに両方のジョブを実行可能な十分なリソース（タスクスロット）があることを確認してください。リソースが不足している場合、このクエリは実行されません。
SELECT kind, SUM(price) FROM test_tbl GROUP BY kind;

ステップ 5：Paimon での OLAP クエリの実行

次の Flink SQL ステートメントを実行して、作成済みのテーブルに対して OLAP クエリを実行します。

-- 実行モードをバッチに設定します。
RESET 'execution.checkpointing.interval';
SET 'execution.runtime-mode' = 'batch';

-- tableau モードを使用して、結果を端末に直接出力します。
SET 'sql-client.execution.result-mode' = 'tableau';

-- テーブル内のデータをクエリします。
SELECT kind, SUM(price) FROM test_tbl GROUP BY kind;

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

重要

テスト終了後は、リソースリークを防ぐため、Paimon のストリーミング書き込みジョブを停止してください。

ジョブを停止した後、次の Flink SQL ステートメントを実行して作成済みのテーブルを削除します。

DROP TABLE test_tbl;