Realtime Compute for Apache Flink:リアルタイムログのデータインジェストによるクイックスタート

アーキテクチャ概要

このパイプラインは、以下のデータフローに従います：

[Faker コネクタ] → [ApsaraMQ for Kafka] → [Realtime Compute for Apache Flink] → [Hologres]

Faker コネクタは合成ユーザー記録を生成し、それを ApsaraMQ for Kafka のトピックへ書き込みます。Realtime Compute for Apache Flink はそのトピックからデータを読み取り、Flink SQL を用いて変換を行い、結果を Hologres データウェアハウスへ書き込みます。これらの 3 つのサービスは、すべて同一の VPC 内に配置する必要があります。

本チュートリアルの手順

1. IP アドレスホワイトリストの設定

2. ApsaraMQ for Kafka インスタンス向けテストデータの準備

3. Hologres カタログの作成（CTAS メソッドのみ）

4. データ同期ジョブの開発および開始

5. 完全データ同期の結果の確認

6. テーブルスキーマ変更の自動同期機能の確認

前提条件

開始する前に、以下の条件が満たされていることを確認してください：

• Realtime Compute for Apache Flink の開発コンソールにアクセスするために使用する Resource Access Management (RAM) ユーザーまたは RAM ロールには、必要な権限が付与されている必要があります。詳細については、「権限管理」をご参照ください。

• Flink ワークスペースが作成されています。詳細については、「Realtime Compute for Apache Flink を有効化する」をご参照ください。

• 上流および下流のストレージインスタンスが作成済みであること：

  • ApsaraMQ for Kafka インスタンスが作成されています。

  • Hologres インスタンスが作成されます。

ステップ 1：IP アドレスホワイトリストの設定

Flink ワークスペースから Kafka インスタンスおよび Hologres インスタンスへアクセスできるようにするため、Flink ワークスペースが配置されている vSwitch の CIDR ブロックを、両インスタンスのホワイトリストに追加します。

1. Flink ワークスペースが配置されている vSwitch の CIDR ブロックを取得します。

   1. Realtime Compute for Apache Flink コンソール にログインします。

   2. 対象のワークスペースを見つけ、[アクション] 列で [その他] > [ワークスペースの詳細] を選択します。

   3. ワークスペースの詳細 ダイアログボックスで、vSwitch の CIDR ブロック をコピーします。

   

2. Kafka インスタンスの IP アドレスのホワイトリストに CIDR ブロックを追加する。

   CIDR ブロックを、[VPC] ネットワークのエンドポイントの許可リストに追加します。

   

3. Hologres インスタンスの IP アドレスホワイトリストに CIDR ブロックを追加する。

   

ステップ 2：ApsaraMQ for Kafka インスタンス向けテストデータの準備

Realtime Compute for Apache Flink の Faker コネクタ をデータ生成ツールとして使用し、データを ApsaraMQ for Kafka インスタンスに書き込みます。

1. ApsaraMQ for Kafka コンソールで、users という名前のトピックを作成します。

2. Kafka トピックへデータを書き込むジョブを開発します。

   1. Realtime Compute for Apache Flink の管理コンソール にログインします。

   2. 対象のワークスペースを見つけ、[コンソール] を [操作] 列でクリックします。

   3. 左側ナビゲーションウィンドウで、開発 > ETL を選択します。表示されたページで、新規作成 をクリックします。

   4. 新規ドラフト ダイアログボックスで、空のストリームドラフト を選択し、次へ をクリックします。ドラフトを以下のように設定します：

      設定項目	例	説明
      名前	kafka-data-input	SQL ドラフトの名前。注：ドラフト名は、現在の名前空間内で一意である必要があります。
      場所	開発	ドラフトのコードファイルが保存されるフォルダ。デフォルトでは、ドラフトのコードファイルは 開発 フォルダに保存されます。既存のフォルダの右側にあるアイコンをクリックすることで、サブフォルダを作成することもできます。
      エンジンバージョン	vvr-8.0.11-flink-1.17	ドラフトのエンジンバージョンをドロップダウンリストから選択します。

   5. 作成 をクリックします。

   6. 以下のコードスニペットを SQL エディターにコピー＆ペーストし、properties.bootstrap.servers のプレースホルダー値を実際の Kafka エンドポイント（上記の注を参照）に置き換えます。

      CREATE TEMPORARY TABLE source ( id INT, first_name STRING, last_name STRING, `address` ROW<`country` STRING, `state` STRING, `city` STRING>, event_time TIMESTAMP ) WITH ( 'connector' = 'faker', 'number-of-rows' = '100', 'rows-per-second' = '10', 'fields.id.expression' = '#{number.numberBetween ''0'',''1000''}', 'fields.first_name.expression' = '#{name.firstName}', 'fields.last_name.expression' = '#{name.lastName}', 'fields.address.country.expression' = '#{Address.country}', 'fields.address.state.expression' = '#{Address.state}', 'fields.address.city.expression' = '#{Address.city}', 'fields.event_time.expression' = '#{date.past ''15'',''SECONDS''}' ); CREATE TEMPORARY TABLE sink ( id INT, first_name STRING, last_name STRING, `address` ROW<`country` STRING, `state` STRING, `city` STRING>, `timestamp` TIMESTAMP METADATA ) WITH ( 'connector' = 'kafka', 'properties.bootstrap.servers' = 'alikafka-host1.aliyuncs.com:9092,alikafka-host2.aliyuncs.com:9092,alikafka-host3.aliyuncs.com:9092', 'topic' = 'users', 'format' = 'json' ); INSERT INTO sink SELECT * FROM source;

      設定項目	例	説明
      properties.bootstrap.servers	alikafka-host1.aliyuncs.com:9092,alikafka-host2.aliyuncs.com:9092,alikafka-host3.aliyuncs.com:9092	Kafka ブローカーのエンドポイント。形式： host:port,host:port,host:port。取得方法については、上記の注を参照してください。
      topic	users	Kafka トピックの名前。

3. ジョブを開始します。

   1. SQL エディターの右上隅にある デプロイ をクリックします。

   2. ドラフトのデプロイ ダイアログボックスで、確認 をクリックします。

   3. ジョブのリソースを構成する。

   4. 「O&M」>「デプロイメント」に移動し、対象のデプロイメントを見つけ、「開始」列の「操作」をクリックします。デプロイメントを開始する際に設定する必要があるパラメーターについて詳しくは、「デプロイメントの開始」をご参照ください。

   5. デプロイメント ページで、デプロイメントの状態を確認します。

      

      Faker コネクタは有限ストリームを提供するため、デプロイメントは RUNNING 状態に達してから約 1 分後に FINISHED 状態に遷移します。終了後、データは送信先の Kafka トピックへ書き込まれます。以下は、ApsaraMQ for Kafka へ書き込まれた JSON 形式のメッセージのサンプルです：

      { "id": 765, "first_name": "Barry", "last_name": "Pollich", "address": { "country": "United Arab Emirates", "state": "Nevada", "city": "Powlowskifurt" } }

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

CTAS を使用した単一テーブル同期の場合、Realtime Compute for Apache Flink の開発コンソールで、送信先カタログとして Hologres カタログを作成します。このセクションでは、基本的な設定項目について説明します。詳細については、「Hologres カタログの作成」をご参照ください。

ステップ 4：データ同期ジョブの開発および開始

ステップ 5：完全データ同期の結果の確認

同期ジョブが RUNNING 状態に達した後、データが Hologres へ書き込まれていることを確認します。

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

2. インスタンス ページで、対象インスタンスの名前をクリックします。

3. ページの右上隅で、インスタンスへの接続 をクリックします。

4. メタデータ管理 タブで、sync_kafka_users テーブルのスキーマおよびデータを確認します。

   

   • テーブルスキーマ：sync_kafka_users テーブルの名前をダブルクリックして、テーブルスキーマを表示します。

     
     説明

     Kafka の partition および offset フィールドを Hologres テーブルのプライマリキーとして宣言します。デプロイメントのフェールオーバーによりデータが再送信された場合でも、同じ partition および offset 値を持つデータは 1 つのコピーのみが保存されます。

   • テーブルデータ：sync_kafka_users テーブルページの右上隅で、テーブルの照会 をクリックします。SQL エディターで以下のステートメントを実行し、実行 をクリックします：

     SELECT * FROM public.sync_kafka_users order by partition, "offset";

     100 行のデータが表示され、Faker コネクタの 'number-of-rows' = '100' 設定と一致しているはずです。

     

ステップ 6：テーブルスキーマ変更の自動同期機能の確認

本ステップでは、スキーマ進化を実証します。新しいネストカラムを含む Kafka メッセージを送信し、Hologres が対応するカラムを自動的に追加することを確認します。

1. ApsaraMQ for Kafka コンソールで、新しいカラムを含むメッセージを送信します。

   1. ApsaraMQ for Kafka コンソール にログインします。

   2. インスタンス ページで、対象インスタンスの名前をクリックします。

   3. 左側ナビゲーションウィンドウで、トピック をクリックします。users という名前のトピックを特定します。

   4. [メッセージの送信] を [操作] 列でクリックします。

   5. メッセージの送信および消費の開始 パネルで、以下のパラメーターを設定します。

      

      設定項目	例
      送信方法	コンソール を選択します。
      メッセージキー	flinktest を入力します。
      メッセージコンテンツ	メッセージコンテンツ フィールドに、以下の JSON コンテンツをコピー＆ペーストします。注：本メッセージには、新しいネストカラム house-points が含まれています。
      指定パーティションへの送信	はい を選択します。
      パーティション ID	0 を入力します。

      { "id": 100001, "first_name": "Dennise", "last_name": "Schuppe", "address": { "country": "Isle of Man", "state": "Montana", "city": "East Coleburgh" }, "house-points": { "house": "Pukwudgie", "points": 76 } }

   6. OK をクリックします。

2. Hologres コンソールで、sync_kafka_users テーブルのスキーマおよびデータの変更を確認します。

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

   2. インスタンス ページで、対象インスタンスの名前をクリックします。

   3. ページの右上隅で、インスタンスへの接続 をクリックします。

   4. メタデータ管理 タブで、sync_kafka_users テーブルの名前をダブルクリックします。

   5. sync_kafka_users テーブルページの右上隅で、テーブルの照会 をクリックします。SQL エディターで以下のステートメントを実行し、実行 をクリックします：

      SELECT * FROM public.sync_kafka_users order by partition, "offset";

   6. テーブルデータを確認します。

      

      結果より、id が 100001 のデータレコードが Hologres へ書き込まれていることがわかります。また、house-points.house および house-points.points のカラムも Hologres へ自動的に追加されています。

      説明

      Kafka の users テーブルの WITH 句で json.infer-schema.flatten-nested-columns.enable が true に設定されているため、Realtime Compute for Apache Flink は、新たに出現したネストカラムを自動的に展開します。各カラムへのアクセスパスが、そのカラム名となります。

まとめおよび次のステップ

Realtime Compute for Apache Flink を使用して、ApsaraMQ for Kafka から Hologres へユーザー記録をリアルタイムでストリーミングするログインジェストパイプラインを、自動スキーマ進化機能付きで正常に構築しました。

本チュートリアルでは、以下の作業を行いました：

• ネットワークホワイトリストを構成し、Flink が Kafka および Hologres へアクセスできるようにしました。

• Faker をベースとしたデータ生成器をデプロイし、100 件の合成レコードを Kafka トピックへ書き込みました。

• Kafka から Hologres へデータを継続的に同期する Flink SQL ジョブを作成しました。

• スキーマ変更（新しいネストカラム）が Hologres へ自動的に伝播されることを検証しました。

さらに学習を進めるには、以下のトピックをご参照ください：

• CREATE TABLE AS (CTAS) 文 — 高度な CTAS オプションおよび動作について学びます。

• Kafka コネクタ — 本番ワークロード向けのコネクタの全構成オプションを確認します。

• デプロイメントを設定する — 本番稼働用にリソース割り当て、チェックポイント、およびその他のデプロイメント設定を調整します。

リファレンス

• CREATE TABLE AS (CTAS) 文

• Kafka コネクタ

• タスクのパフォーマンスを向上させるためにノードの並列度およびリソースを変更する方法について詳しくは、「デプロイメントの設定」をご参照ください。