Realtime Compute for Apache Flink:リアルタイムログ取り込み

前提条件

• ご利用の RAM ユーザーまたは RAM ロールが Realtime Compute for Apache Flink コンソールへのアクセスに必要な権限を持っていることを確認してください。詳細については、「権限」をご参照ください。

• Flink ワークスペースを作成しました。詳細については、「ワークスペースを作成する」をご参照ください。

• アップストリームおよびダウンストリームのストレージ

  • ApsaraMQ for Kafka インスタンスを作成済みである必要があります。詳細については、「ステップ 2：インスタンスの購入とデプロイ」をご参照ください。

  • Hologres インスタンスを作成済みである必要があります。詳細については、「Hologres インスタンスの購入」をご参照ください。

  説明

  ご利用の ApsaraMQ for Kafka および Hologres インスタンスは、Realtime Compute for Apache Flink ワークスペースと同じリージョンおよび VPC 内にある必要があります。そうでない場合は、ネットワーク接続を確立する必要があります。詳細については、「VPC 間でのサービスへのアクセス」または「インターネットへのアクセス」をご参照ください。

ステップ 1：IP ホワイトリストの構成

Flink がご利用の Kafka および Hologres インスタンスにアクセスできるようにするため、Flink ワークスペースの CIDR ブロックを Kafka および Hologres の IP ホワイトリストに追加します。

1. Flink ワークスペースの VPC CIDR ブロックを取得します。

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

   2. 対象の[ワークスペース]の[操作]列で、[その他] > [ワークスペースの詳細]を選択します。

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

      ダイアログボックスには、ワークスペースの基本情報および VSwitch の一覧が表示されます。CIDR ブロック列で、各アベイラビリティゾーンの CIDR ブロックを確認し、次のステップで使用するためにメモしておいてください。

2. Flink ワークスペースの CIDR ブロックを、ご利用の Kafka インスタンスの IP ホワイトリストに追加します。

   VPC エンドポイントの IP ホワイトリストを構成する必要があります。詳細な手順については、「IP ホワイトリストの構成」をご参照ください。ホワイトリスト編集ダイアログボックスで、ホワイトリスト IP の追加をクリックして CIDR ブロックを追加します。

3. Flink ワークスペースの CIDR ブロックを、ご利用の Hologres インスタンスの IP ホワイトリストに追加します。

   Hologres インスタンスにログインし、その IP ホワイトリストを構成します。詳細な手順については、「IP ホワイトリスト」をご参照ください。HoloWeb のセキュリティセンターにあるホワイトリスト構成ページで、IP ホワイトリストの編集ダイアログボックスのIP アドレスフィールドに CIDR ブロックを入力し、OKをクリックします。

ステップ 2：Kafka テストデータの準備

Realtime Compute for Apache Flink のFakerコネクタを使用して、ApsaraMQ for Kafka にデータを生成および書き込みます。Realtime Compute for Apache Flink の開発コンソールで以下の手順に従ってください。

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

   詳細については、「トピックの作成」をご参照ください。

2. ApsaraMQ for Kafka にデータを書き込むジョブを作成します。

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

   2. ターゲットワークスペースの操作列で、コンソールをクリックします。

   3. 左側のナビゲーションウィンドウで、開発 > ETLを選択します。

   4. アイコンをクリックし、次にストリームドラフトの新規作成をクリックします。ファイル名を入力し、エンジンバージョンを選択します。

      Realtime Compute for Apache Flink には、さまざまなコードテンプレートおよびデータ同期テンプレートも用意されています。各テンプレートには、特定のユースケース、コードサンプル、および説明が含まれています。テンプレートをクリックすると、Realtime Compute for Apache Flink の機能や構文を素早く学習し、ビジネスロジックを実装できます。詳細については、「コードテンプレート」および「データ同期テンプレート」をご参照ください。

      パラメーター	説明	例
      ファイル名	ジョブの名前です。 説明 ジョブ名は、現在のワークスペース内で一意である必要があります。	flink-test
      エンジンバージョン	現在のジョブの Flink エンジンバージョンです。 信頼性とパフォーマンスを高めるために、推奨または安定版とラベル付けされたバージョンを選択してください。エンジンバージョンの詳細については、「リリースノート」および「エンジンバージョン」をご参照ください。	vvr-8.0.8-flink-1.17

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

   6. ジョブの SQL ステートメントを記述します。

      以下のコードをエディターにコピーし、ご利用の環境に合わせてパラメーターを変更してください。

      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',
        'properties.enable.idempotence'='false'
      );
      
      INSERT INTO sink SELECT * FROM source;

      以下の表は、変更が必要なパラメーターについて説明しています。

      パラメーター	例の値	説明
      properties.bootstrap.servers	alikafka-host1.aliyuncs.com:9092,alikafka-host2.aliyuncs.com:9092,alikafka-host3.aliyuncs.com:9092	ApsaraMQ for Kafka ブローカーエンドポイントです。 'host:port' 形式のブローカーエンドポイントをカンマ区切りで指定します。ドメイン名エンドポイント (VPC) は、インスタンスの詳細ページのエンドポイント情報セクションで確認できます。
      topic	users	ApsaraMQ for Kafka トピック名です。

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

   1. 開発 > ETLページで、デプロイをクリックします。

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

   3. ジョブのリソースを構成します。詳細については、「ジョブのリソース構成」をご参照ください。

   4. 運用管理 > デプロイメントページで、ターゲットのデプロイメントを検索し、操作列の開始をクリックします。起動構成の詳細については、「デプロイメントの開始」をご参照ください。

   5. デプロイメントページで、デプロイメントのランタイム情報およびステータスをモニターできます。

      Faker ソースは有界ストリームを生成するため、デプロイメントのステータスは開始から約 1 分後に FINISHED に変化します。デプロイメントが完了すると、データは 'users' トピックに書き込まれます。以下のコードは、JSON 形式のデータのサンプルです。

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

ステップ 3：データ同期ジョブの作成と開始

ステップ 4：完全同期結果の確認

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

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

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

4. メタデータ管理タブで、flink_test_db データベース内の同期済み users テーブルのテーブルスキーマおよびデータを確認します。

   左側のナビゲーションツリーで、インスタンス名 > flink_test_db > test_schema > テーブルを展開します。同期済みのusersテーブルが表示されます。

   同期済みテーブルのスキーマおよびデータは以下のとおりです。

   • テーブルスキーマ

     users テーブル名をダブルクリックして、テーブルスキーマを確認します。

     users テーブルスキーマには、以下のフィールドが含まれます：id (BIGINT、プライマリキー)、first_name (TEXT)、last_name (TEXT)、address.country (TEXT)、address.state (TEXT)、およびaddress.city (TEXT)。

     説明

     完全データ同期中は、ジョブがフェイルオーバーしてデータを再送信した場合にデータ重複を防ぐため、Kafka メタデータの partition および offset を Hologres テーブルのプライマリキーとして定義することを推奨します。

   • テーブルデータ

     users テーブルページの右上隅で、テーブルのクエリをクリックします。以下のコマンドを入力し、実行をクリックします。

     SELECT * FROM test_schema.users;

     コマンドは以下の結果を返します。

     クエリは複数行を返します。これにより、レコードが users テーブルに正常に同期されたことが確認できます。返された各行には、id、first_name、last_name、address.country、address.state、およびaddress.cityカラムの完全なデータが含まれています。

ステップ 5：自動スキーマ同期の観察

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

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

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

   3. トピックページで、ターゲットトピック (users) の名前をクリックします。

   4. メッセージの送信をクリックします。

   5. メッセージを構成します。

      メッセージの送信と消費の開始ダイアログボックスで、以下のとおりパラメーターを構成します。

      パラメーター	例
      送信方法	コンソールを選択します。
      メッセージキー	flinktest と入力します。
      メッセージ内容	以下の JSON コンテンツをコピーして、メッセージ内容フィールドに貼り付けます。 { "id": 100001, "first_name": "Dennise", "last_name": "Schuppe", "address": { "country": "Isle of Man", "state": "Montana", "city": "East Coleburgh" }, "house-points": { "house": "Pukwudgie", "points": 76 } } 説明 この例では、house-points が新しいネストされたカラムです。
      指定パーティションに送信	はいを選択します。
      パーティション ID	0 と入力します。

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

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

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

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

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

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

   5. テーブルのクエリをクリックし、以下のステートメントを入力して、実行をクリックします。

      SELECT * FROM test_schema.users;

   6. クエリ結果を確認します。

      クエリは以下の結果を返します。

      結果から、ID 100001 のレコードが Hologres に正常に書き込まれ、house-points.house および house-points.points の 2 つの新しいカラムが Hologres テーブルに追加されたことがわかります。

      説明

      ApsaraMQ for Kafka に送信されたメッセージには、ネストされたカラムが house-points のみ含まれています。ただし、WITH 句でjson.infer-schema.flatten-nested-columns.enableが指定されているため、Realtime Compute for Apache Flink がこのカラムを自動的にフラット化し、ネストされたフィールドのアクセスパスを新しいカラム名として使用します。

参照

• CREATE TABLE AS (CTAS) 文の詳細については、「CREATE TABLE AS (CTAS) 文」をご参照ください。

• Message Queue for Apache Kafka をソースまたは結果テーブルとして使用する方法については、「Message Queue for Apache Kafka」をご参照ください。

• ノードの並列度およびリソースを調整してジョブのパフォーマンスを向上させる方法については、「ジョブデプロイの構成」をご参照ください。