Realtime Compute for Apache Flink:Simple Log Service (SLS)

構文

CREATE TABLE sls_table(
  a INT,
  b INT,
  c VARCHAR
) WITH (
  'connector' = 'sls',
  'endPoint' = '<yourEndPoint>',
  'project' = '<yourProjectName>',
  'logStore' = '<yourLogStoreName>',
  'accessId' = '${secret_values.ak_id}',
  'accessKey' = '${secret_values.ak_secret}'
);

WITH パラメーター

• 共通パラメーター

  パラメーター	説明	データ型	必須	デフォルト値	注意
  connector	テーブルタイプ。	String	はい	なし	固定値：sls。
  endPoint	エンドポイントアドレス。	String	はい	なし	SLS のプライベートネットワークエンドポイントを入力します。詳細については、「サービスエンドポイント」をご参照ください。 説明 Realtime Compute for Apache Flink は、デフォルトではパブリックネットワークアクセスをサポートしていません。ただし、Alibaba Cloud NAT Gateway を使用すると、VPC ネットワークとパブリックネットワーク間の通信が可能になります。詳細については、「パブリックネットワークへのアクセス方法」をご参照ください。 パブリックネットワーク経由での SLS へのアクセスは避けてください。必要な場合は、HTTPS を使用し、SLS の Global Accelerator (GA) を有効にしてください。詳細については、「転送アクセラレーションの管理」をご参照ください。
  project	SLS プロジェクト名。	String	はい	なし	なし。
  logStore	SLS Logstore またはメトリックストア名。	String	はい	なし	Logstore とメトリックストアは同じ消費メソッドを使用します。
  accessId	ご利用の Alibaba Cloud アカウントの AccessKey ID。	String	はい	なし	詳細については、「AccessKey ID と AccessKey Secret の表示方法」をご参照ください。 重要 AccessKey 情報の漏洩を防ぐため、変数を使用して AccessKey の値を指定してください。詳細については、「プロジェクト変数」をご参照ください。
  accessKey	ご利用の Alibaba Cloud アカウントの AccessKey Secret。	String	はい	なし

• ソース固有のパラメーター

  パラメーター	説明	データ型	必須	デフォルト値	注意
  enableNewSource	FLIP-27 インターフェイスを実装する新しいソースを有効にするかどうか。	Boolean	いいえ	false	新しいソースはシャードの変更に自動的に適応し、すべてのソースタスクにシャードを均等に分散します。 重要 このパラメーターは VVR 8.0.9 以降でサポートされています。VVR 11.1 以降、デフォルト値は true です。 このパラメーターを変更すると、ジョブは以前の状態から再開できません。過去のオフセットから消費を再開するには、まず consumerGroup パラメーターを指定してジョブを開始し、SLS コンシューマーグループに消費の進行状況を記録します。次に、consumeFromCheckpoint を true に設定し、状態なしでジョブを再起動します。 SLS に読み取り専用シャードが存在する場合、一部の Flink タスクは読み取り専用シャードの消費が完了した後も、他の未処理のシャードをリクエストし続けることがあります。これにより、同時実行タスク間でシャードの分散が不均一になり、全体的な消費効率とシステムパフォーマンスが低下する可能性があります。この問題を軽減するには、並列度を調整するか、タスクスケジューリングを最適化するか、または小さなシャードをマージしてシャード数とタスク割り当ての複雑さを軽減します。
  shardDiscoveryIntervalMs	シャードの変更を動的に検出する間隔 (ミリ秒)。	Long	いいえ	60000	負の値を設定すると、動的検出が無効になります。 説明 この値は 60000 ms (1 分) 以上である必要があります。 このパラメーターは、enableNewSource が true の場合にのみ有効です。 このパラメーターは VVR 8.0.9 以降でサポートされています。
  startupMode	ソーステーブルの起動モード。	String	いいえ	timestamp	timestamp (デフォルト)：指定された時刻からログの消費を開始します。 latest：最新のオフセットからログの消費を開始します。 earliest：最も早いオフセットからログの消費を開始します。 consumer_group：コンシューマーグループに記録されたオフセットからログの消費を開始します。シャードにオフセットが記録されていない場合、消費は最も早いオフセットから開始されます。 重要 VVR 11.1 より前のバージョンは consumer_group をサポートしていません。consumeFromCheckpoint を true に設定してください。消費は指定されたコンシューマーグループに記録されたオフセットから開始され、startupMode は効果がありません。
  startTime	ログ消費の開始時間。	String	いいえ	現在の時刻	フォーマット：yyyy-MM-dd hh:mm:ss。 startupMode が timestamp に設定されている場合にのみ有効です。 説明 startTime と stopTime は、__timestamp__ ではなく、SLS の __receive_time__ 属性に基づいています。
  stopTime	ログ消費の終了時間。	String	いいえ	なし	フォーマット：yyyy-MM-dd hh:mm:ss。 説明 このパラメーターは、過去のログのみを消費する場合に使用します。過去の時刻に設定してください。未来の時刻に設定すると、新しいログがないために消費が早期に終了し、エラーメッセージなしでデータストリームが中断される可能性があります。 ログ消費が完了したときに Flink プログラムを終了するには、exitAfterFinish=true も設定する必要があります。
  consumerGroup	コンシューマーグループ名。	String	いいえ	なし	コンシューマーグループは消費の進行状況を記録します。制限なくカスタム名を定義できます。 説明 同じコンシューマーグループを使用して複数のジョブ間で消費を調整することはできません。各 Flink ジョブは一意のコンシューマーグループを使用する必要があります。複数のジョブが同じコンシューマーグループを共有すると、すべてのデータを消費してしまいます。これは、Flink が SLS コンシューマーグループを介してパーティションを割り当てないため、各コンシューマーが共有グループに関係なくメッセージを独立して処理するためです。
  consumeFromCheckpoint	指定されたコンシューマーグループに保存されているチェックポイントからログの消費を開始するかどうか。	String	いいえ	false	true：コンシューマーグループを指定します。Flink プログラムは、そのグループに保存されているチェックポイントからログの消費を開始します。チェックポイントが存在しない場合、消費は startTime の値から開始されます。 false (デフォルト)：指定されたコンシューマーグループに保存されているチェックポイントからログの消費を開始しません。 重要 VVR 11.1 以降では、このパラメーターはサポートされていません。VVR 11.1 以降では、startupMode を consumer_group に設定してください。
  maxRetries	SLS の読み取りに失敗した後の再試行回数。	String	いいえ	3	なし。
  batchGetSize	リクエストごとに読み取るロググループの数。	String	いいえ	100	batchGetSize の値は 1000 を超えてはなりません。超えた場合、エラーが発生します。
  exitAfterFinish	データ消費が完了した後に Flink プログラムを終了するかどうか。	String	いいえ	false	true：データ消費が完了した後に Flink プログラムを終了します。 false (デフォルト)：データ消費が完了した後も Flink プログラムは実行を継続します。
  query 重要 VVR 11.3 以降では非推奨です。以降のバージョンでも互換性は維持されます。	SLS 消費の前処理ステートメント。	String	いいえ	なし	query パラメーターを使用して、消費前に SLS データをフィルターします。これにより、すべてのデータを Flink に消費することを避け、コストを節約し、処理速度を向上させます。 たとえば、 'query' = '*| where request_method = ''GET''' は、Flink が読み取る前に request_method フィールドが GET に等しいログに一致します。 説明 クエリには SPL 構文を使用します。詳細については、「SPL 構文」をご参照ください。 重要 このパラメーターは VVR 8.0.1 以降でサポートされています。 この機能には SLS の料金が発生します。詳細については、「料金」をご参照ください。
  processor	SLS コンシューマープロセッサ。query と両方が存在する場合、こちらが優先されます。	String	いいえ	なし	processor パラメーターを使用して、消費前に SLS データをフィルターします。これにより、すべてのデータを Flink に消費することを避け、コストを節約し、処理速度を向上させます。query よりも processor の使用を推奨します。 たとえば、 'processor' = 'test-filter-processor' は、Flink がデータを読み取る前に SLS コンシューマープロセッサを適用します。 説明 プロセッサには SPL 構文を使用します。詳細については、「SPL 構文」をご参照ください。コンシューマープロセッサの作成または更新の手順については、「コンシューマープロセッサの管理」をご参照ください。 重要 このパラメーターは VVR 11.3 以降でサポートされています。 この機能には SLS の料金が発生します。詳細については、「料金」をご参照ください。

• 結果テーブルのみ

  パラメーター	説明	データ型	必須	デフォルト値	注意
  topicField	値が __topic__ メタデータフィールドをオーバーライドするフィールドの名前。ログトピックを表します。	String	いいえ	なし	このフィールドはテーブルに存在する必要があります。
  timeField	値が __timestamp__ メタデータフィールドをオーバーライドするフィールドの名前。ログの書き込み時間を表します。	String	いいえ	現在の時刻	このフィールドはテーブルに存在し、データ型が INT である必要があります。指定しない場合、現在の時刻が使用されます。
  sourceField	値が __source__ メタデータフィールドをオーバーライドするフィールドの名前。ログを生成したマシンの IP アドレスなどのログソースを表します。	String	いいえ	なし	このフィールドはテーブルに存在する必要があります。
  partitionField	フィールドの名前。データはこのフィールドのハッシュ値に基づいてシャードに書き込まれます。同じハッシュ値を持つデータは同じシャードに書き込まれます。	String	いいえ	なし	指定しない場合、各レコードは利用可能なシャードにランダムに書き込まれます。
  buckets	partitionField が指定された場合にデータを再割り当てするためのバケット数。	String	いいえ	64	有効な値：1 から 256 までの 2 のべき乗の整数。バケット数はシャード数以上である必要があります。そうでない場合、一部のシャードはデータを受信しません。
  flushIntervalMs	データ書き込みをトリガーする時間間隔。	String	いいえ	2000	単位：ミリ秒。
  writeNullProperties	null 値を空文字列として SLS に書き込むかどうか。	Boolean	いいえ	true	true (デフォルト)：null 値を空文字列として書き込みます。 false：書き込み中に null 値を持つフィールドをスキップします。 説明 このパラメーターは VVR 8.0.6 以降でサポートされています。

型マッピング

制限事項

この機能は VVR 11.1 以降でのみサポートされています。

構文

source:
   type: sls
   name: SLS Source
   endpoint: <endpoint>
   project: <project>
   logstore: <logstore>
   accessId: <accessId>
   accessKey: <accessKey>

設定項目

型マッピング

fixed-types が設定されていない場合、データインジェストの型マッピングは次のようになります：

fixed-types が設定されている場合、コネクタは指定された型を使用してデータの解析を試みます。

スキーマ推論とスキーマ変更の同期

• シャードの事前消費とスキーマの初期化

  SLS コネクタは、データを読み取る Logstore の現在のスキーマを維持します。データを読み取る前に、シャードごとに最大 `maxPreFetchLogGroups` 個のロググループを事前消費します。その後、各ログのスキーマを解析し、これらのスキーマをマージしてテーブル構造を初期化します。続いて、実際の消費の前に、初期化されたスキーマに基づいて対応するテーブル作成イベントを生成します。

  説明

  各シャードについて、SLS コネクタは現在の時刻の 1 時間前からログスキーマの消費と解析を試みます。

• プライマリキー情報

  SLS ログにはプライマリキー情報が含まれていません。変換ルールを使用して手動でプライマリキーを追加できます：

  transform:
    - source-table: <project>.<logstore>
      projection: \*
      primary-keys: key1, key2

• スキーマ推論とスキーマ変更

  スキーマの初期化後、schema.inference.strategy が static に設定されている場合、SLS コネクタは初期スキーマを使用して各ログを解析し、スキーマ変更イベントを生成しません。schema.inference.strategy が continuous に設定されている場合、SLS コネクタは各ログを解析し、物理列を推論し、それらを現在のスキーマと比較します。その後、スキーマが異なる場合にスキーマをマージします。マージルール：

  • 推論された物理列に現在のスキーマに存在しないフィールドが含まれている場合、それらのフィールドはスキーマに追加され、新しい null 値許容列イベントが生成されます。

  • 推論された物理列に現在のスキーマに既に存在するフィールドが含まれていない場合、それらのフィールドは残ります。そのデータは NULL で埋められ、列削除イベントは生成されません。

  SLS コネクタは、ログ内のすべてのフィールド型を `STRING` として推論します。現在、列の追加のみがサポートされています。新しい列は現在のスキーマの末尾に追加され、null 値許容として設定されます。