Flink DataStream API を使用した DLF へのデータインジェスト - Data Lake Formation

このトピックでは、DataStream API を使用して Paimon REST 経由で Data Lake Formation (DLF) カタログにデータを書き込む Realtime Compute for Apache Flink ジョブを開発する方法について説明します。

前提条件

Realtime Compute for Apache Flink のフルマネージドワークスペースが必要です。作成していない場合は、「Realtime Compute for Apache Flink の有効化」をご参照ください。
DLF カタログが作成されていること。詳細については、「DLF の使用開始」をご参照ください。
お使いの Flink ワークスペースと DLF カタログが同じリージョンに存在すること。
お使いの Flink ワークスペースの VPC が DLF の VPC ホワイトリストに含まれていること。詳細については、「VPC ホワイトリストの設定」をご参照ください。
説明
DLF はデフォルトで VPC アクセスを有効にします。パブリックネットワークアクセスを有効にするには、「パブリックネットワークアクセスの有効化と設定」をご参照ください。

事前準備

Apache Paimon の Web サイトから、バージョン 1.1 以降の Paimon バンドル JAR paimon-flink-*.jar をダウンロードします。
Apache Paimon Filesystems から、バージョン 1.1 以降の paimon-oss-*.jar をダウンロードします。

依存関係の追加方法

重要

Flink ランタイム環境には、Paimon コネクタや OSS ファイルシステムは含まれていません。ジョブの実行時に paimon-flink-*.jar と paimon-oss-*.jar を利用可能にするには、次のいずれかの方法を使用する必要があります。

方法 1：コンソールでの追加ファイルのアップロード

pom.xml ファイルを変更する必要はありません。Realtime Compute for Apache Flink 開発コンソールで JAR ジョブを作成するときは、準備セクションでダウンロードした paimon-flink-*.jar および paimon-oss-*.jar ファイルを追加の依存ファイルとしてアップロードします。

方法 2：Maven を使用した Fat JAR への依存関係のパッケージ化

以下の依存関係とプロパティをプロジェクトの pom.xml ファイルに追加します。

<properties>
    <!-- Paimon のバージョン。1.1 以降を指定します。 -->
    <paimon.version>1.1.0</paimon.version>
    <!-- Flink のメジャーバージョン。VVR バージョンに基づいて設定します。次の表をご参照ください。 -->
    <flink.main.version>1.20</flink.main.version>
</properties>

<dependencies>
    <dependency>
        <groupId>org.apache.paimon</groupId>
        <artifactId>paimon-flink-${flink.main.version}</artifactId>
        <version>${paimon.version}</version>
    </dependency>

    <dependency>
        <groupId>org.apache.paimon</groupId>
        <artifactId>paimon-oss</artifactId>
        <version>${paimon.version}</version>
    </dependency>
</dependencies>

${flink.main.version} の値は以下のとおりです。

VVR バージョン	`flink.main.version`
VVR 8.x	1.17
VVR 11.x	1.20

この方法を使用すると、依存関係が Fat JAR にパッケージ化されるため、デプロイ中に追加の JAR ファイルをアップロードする必要はありません。

ステップ 1：ジョブコードの作成

DataStream ジョブの main() メソッドで、次のコードを使用して DLF カタログインスタンスを作成します。

Options options = new Options();
options.set("type", "paimon");
options.set("metastore", "rest");
options.set("uri", "http://<region-id>-vpc.dlf.aliyuncs.com");
options.set("warehouse", "your-catalog-name");
options.set("token.provider", "dlf");
options.set("dlf.access-key-id", "your-access-key-id");
options.set("dlf.access-key-secret", "your-access-key-secret");
Catalog catalog = CatalogFactory.createCatalog(options);

必須パラメーター：

パラメーター	説明	例
`type`	カタログタイプ。カスタム JAR から自動的に解析されます。この値は変更しないでください。	`paimon`
`metastore`	DLF のメタストアタイプです。`rest` に設定します。	`rest`
`uri`	DLF REST カタログサーバーの VPC エンドポイントです。形式は `http://[region-id]-vpc.dlf.aliyuncs.com` です。詳細については、「リージョンとエンドポイント」をご参照ください。	`http://ap-southeast-1-vpc.dlf.aliyuncs.com`
`warehouse`	Paimon カタログ名。	`dlf_test`
`token.provider`	トークンプロバイダー。 `dlf` に設定します。	`dlf`
`dlf.access-key-id`	認証用の AccessKey ID。詳細については、「RAM ユーザーの AccessKey 情報の表示」をご参照ください。
`dlf.access-key-secret`	認証用の AccessKey Secret。

カタログが作成されたら、DataStream ジョブに登録して使用すると、Paimon テーブルの読み取りと書き込みができます。

ステップ 2：ジョブのパッケージ化とデプロイ

DataStream ジョブを JAR ファイルにパッケージ化します。
Realtime Compute for Apache Flink コンソールでジョブ JAR をアップロードし、ジョブを送信します。
方法 1 (コンソールでの追加ファイルのアップロード) を選択した場合は、ジョブの送信時に追加の依存関係に paimon-flink-*.jar と paimon-oss-*.jar を追加します。

Flink JAR ジョブの開発とデバッグの詳細については、「JAR ジョブの開発」をご参照ください。