Realtime Compute for Apache Flink:Message Service (MNS)

背景情報

Message Service (MNS) は、効率的、高信頼性、安全、使いやすく、弾性的にスケーラブルな分散メッセージングサービスです。MNS は、Object Storage Service (OSS) のイベント通知機能を提供します。イベント通知ルールを作成することで、指定した OSS リソースでオブジェクト作成などのイベントが発生した際に、メッセージを MNS キューにプッシュできます。

Flink ジョブは MNS コネクタを使用してこれらのイベントを消費できます。たとえば、リアルタイム画像処理シナリオでは、MNS コネクタを使用して OSS バケット内の新しいファイルのパスを取得できます。その後、Realtime Compute for Apache Flink が提供する FETCH_CONTENT 関数を使用して画像コンテンツをダウンロードし、AI 大規模言語モデル統合 を使用してリアルタイムのマルチモーダル分析を行うことができます。

前提条件

• MNS を有効化し、必要な権限を付与していること。内部ネットワークアクセスの場合、MNS キューはご利用の Flink ワークスペースと同じリージョンにある必要があります。

• パブリックインターネット経由で MNS にアクセスする場合、ご利用の Flink ワークスペースの パブリックアクセスを有効にし、Flink ワークスペースのパブリック IP アドレスを MNS キューの許可リストに追加する必要があります。詳細については、「MNS アクセス制御」をご参照ください。

制限事項

• MNS コネクタは、Realtime Compute for Apache Flink VVR 11.6.0 以降でのみサポートされています。

• Kafka とは異なり、MNS コネクタは特定のオフセットからのメッセージ消費やシーク操作をサポートしていません。詳細については、「Message Service (MNS)」をご参照ください。

• 並列度 1 に固定： MNS コネクタは、コネクタ側の重複排除によって Exactly-Once セマンティクスを実現し、並列度は 1 のみをサポートします。

• チェックポイントの有効化が必須： MNS コネクタは、メッセージの確認と削除をチェックポイントに依存しています。チェックポイントが有効になっていない場合、メッセージは削除されず、無限に繰り返し消費されることになります。

• メッセージ本文のサイズ制限： 1 つの MNS メッセージの本文は 64 KB を超えることはできません。より大きなメッセージを処理する方法については、MNS ドキュメントの「大きなメッセージの転送」のベストプラクティスをご参照ください。

• 可視性タイムアウトの設定： MNS キューを作成する際には、可視性タイムアウト を設定する必要があります。メッセージの再配信を避けるため、可視性タイムアウトはチェックポイント間隔よりも大きい値に設定してください。

• MNS コネクタは、消費されるイベントの厳密な順序を保証しません。 MNS スタンダードキューは、厳密な先入れ先出し (FIFO) ではありません。タイムアウト後のメッセージ再配信により、消費の順序が変わる可能性があります。

構文

CREATE TABLE mns_source (
    data STRING
) WITH (
    'connector' = 'mns',
    'endpoint' = '${endpoint}',
    'region' = '${region}',
    'queueName' = '${queueName}',
    'accessKeyId' = '${accessKeyId}',
    'accessKeySecret' = '${accessKeySecret}',
    'format' = 'json',
    'batchSize' = '8',
    'pollingWaitTime' = '10s',
    'messageType' = 'RAW'
);

WITH パラメーター

• 一般

  パラメーター	説明	型	必須	デフォルト値	備考
  connector	コネクタのタイプ。	String	はい	なし	値は mns に固定されます。
  endpoint	MNS サービスのエンドポイント。	String	はい	なし	URI のフォーマットは http://{account-id}.mns.{region}.aliyuncs.com です。 詳細については、「リージョンエンドポイント」をご参照ください。
  region	MNS キューのリージョン。	String	はい	なし	例：cn-hangzhou。サポートされているリージョンのリストについては、「エンドポイント」をご参照ください。
  queueName	MNS キュー名。	String	はい	なし	MNS コンソールでキューを作成する際に設定した名前。
  accessKeyId	MNS サービスにアクセスするための AccessKey ID。	String	はい	なし	既存の AccessKey を使用するか、「AccessKey の作成」をご参照ください。
  accessKeySecret	MNS サービスにアクセスするための AccessKey Secret。	String	はい	なし
  format	データ形式。	String	はい	なし	有効な値： csv json avro parquet orc raw
  batchSize	MNS キューから一度にプルするメッセージの最大数。	Integer	いいえ	1	範囲：1～16。これは読み取りパフォーマンスに影響します。値を大きくするとスループットが向上する可能性があります。注意：MNS サービスの API は、1 回の呼び出しで最大 16 メッセージに制限されています。
  pollingWaitTime	MNS キューをポーリングする際のメッセージの最大待機時間。	Duration	いいえ	10s	範囲：0s～30s。値が 0s の場合は待機しません。
  messageType	MNS メッセージのペイロードタイプ。	String	いいえ	RAW	値は RAW または OSS です。 RAW (デフォルト)：メッセージ本文は標準の JSON として扱われ、設定された format デシリアライザーによって解析されます。 OSS：OSS イベント通知 からの JSON データを自動的に解析します。(OSS イベントへの MNS サブスクリプションが必要です。) コネクタは events 配列から最初の要素を自動的に抽出し、ネストされたフィールドをフラット化してテーブル構造にマッピングします。 説明 messageType が OSS に設定されている場合、format は json である必要があります。
  deleteMaxRetries	メッセージの削除が失敗した場合の最大リトライ回数。	Integer	いいえ	3	なし。
  startTimeMs	消費を開始する時刻 (ミリ秒単位の UNIX タイムスタンプ)。	Long	いいえ	-1	値が -1 の場合、タイムスタンプによるフィルタリングは無効になり、コネクタはすべての可視メッセージを消費します。 説明 MNS コネクタは Kafka のようなシークをサポートしていません。startTimeMs パラメーターはフィルタリング専用です。コネクタは、このタイムスタンプより前にキューに入ったメッセージを除外します。

OSS イベント通知の読み取り

例

• 例 1：標準の JSON メッセージを消費する

  -- MNS ソーステーブルを作成します。
  -- フィールド名は JSON メッセージ本文のキーと一致する必要があります。
  CREATE TEMPORARY TABLE mns_source (
    `userId` BIGINT,
    `action` STRING,
    `timestamp` TIMESTAMP(3),
    `payload` STRING
  ) WITH (
    'connector' = 'mns',
    'endpoint' = 'http://your-account-id.mns.cn-hangzhou.aliyuncs.com',
    'region' = 'cn-hangzhou',
    'queueName' = 'my-events-queue',
    'accessKeyId' = 'your-ak',
    'accessKeySecret' = 'your-sk',
    'format' = 'json'
  );
  
  -- 出力をテストするための結果テーブルを作成します。
  CREATE TEMPORARY TABLE print_sink (
    `userId` BIGINT,
    `action` STRING,
    `timestamp` TIMESTAMP(3),
    `payload` STRING
  ) WITH (
    'connector' = 'print'
  );
  
  -- データを消費して出力します。
  INSERT INTO print_sink
  SELECT userId, action, timestamp, payload
  FROM mns_source;

• 例 2：OSS イベント通知メッセージを消費する

  -- OSS イベント通知 JSON を自動的に解析するための MNS ソーステーブルを作成します。
  -- フィールド名の定義については、「OSS イベント通知の読み取り」セクションをご参照ください。
  CREATE TEMPORARY TABLE oss_event_source (
    `eventName` STRING,
    `eventSource` STRING,
    `eventTime` TIMESTAMP(3),
    `region` STRING,
    `ossBucketName` STRING,
    `ossObjectKey` STRING,
    `ossObjectSize` BIGINT,
    `responseElementsRequestId` STRING,
    `userIdentityPrincipalId` STRING
  ) WITH (
    'connector' = 'mns',
    'endpoint' = 'http://123456789.mns.cn-hangzhou.aliyuncs.com',
    'region' = 'cn-hangzhou',
    'queueName' = 'oss-events-queue',
    'accessKeyId' = '${secret_values.ak_id}',
    'accessKeySecret' = '${secret_values.ak_secret}',
    'format' = 'json',
    'messageType' = 'OSS'
  );
  
  -- 出力をテストするための結果テーブルを作成します。
  CREATE TEMPORARY TABLE print_sink (
    `eventName` STRING,
    `ossBucketName` STRING,
    `ossObjectKey` STRING,
    `ossObjectSize` BIGINT,
    `eventTime` TIMESTAMP(3)
  ) WITH (
    'connector' = 'print'
  );
  
  -- ObjectCreated イベントをフィルタリングして結果を出力します。
  INSERT INTO print_sink
  SELECT eventName, ossBucketName, ossObjectKey, ossObjectSize, eventTime
  FROM oss_event_source
  WHERE eventName LIKE 'ObjectCreated:%';