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 の VPC エンドポイントを入力します。詳細については、「エンドポイント」をご参照ください。 説明 デフォルトでは、Realtime Compute for Apache Flink はインターネットにアクセスできません。ただし、Alibaba Cloud では NAT ゲートウェイを提供しており、VPC とインターネット間の通信を有効化できます。詳細については、「インターネットへのアクセス方法」をご参照ください。 SLS へのインターネット経由のアクセスは避けてください。やむを得ずインターネット経由でアクセスする場合は、HTTPS を使用し、SLS の転送アクセラレーションを有効化してください。
  project	SLS プロジェクトの名称です。	String	はい	なし	該当なし。
  logStore	SLS Logstore または Metricstore の名称です。	String	はい	なし	Logstore 内のデータは、Metricstore 内のデータと同じ方法で消費されます。
  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 オプションを指定してジョブを起動し、現在のコンシューマオフセットを記録します。その後、consumeFromCheckpoint を true に設定して、状態なしでジョブを再起動します。 SLS に読み取り専用のシャードが含まれている場合、一部の Flink サブタスクがそれらのシャードからの読み取りを完了した後、他の未読シャードを要求することがあります。これにより、サブタスク間のシャード分散が不均等になり、全体的な消費効率およびシステムパフォーマンスが低下する可能性があります。この不均衡を軽減するには、ソースの並列度を調整するか、タスクスケジューリングを最適化するか、小さなシャードをマージしてください。
  shardDiscoveryIntervalMs	シャードの変更を動的に検出する間隔です（単位：ミリ秒）。	Long	いいえ	60000	このオプションを負の値に設定すると、動的検出が無効になります。 説明 このオプションの最小値は 1 分（60,000 ミリ秒）です。 このオプションは、enableNewSource が true に設定されている場合にのみ有効です。 このオプションは VVR 8.0.9 以降でのみサポートされます。
  startupMode	ソーステーブルの起動モードです。	String	いいえ	timestamp	timestamp（デフォルト）：指定された時刻からログを消費します。 latest：最新のオフセットからログを消費します。 earliest：最も古いオフセットからログを消費します。 consumer_group：コンシューマグループに記録されたオフセットからログを消費します。シャードに対してオフセットが記録されていない場合は、最も古いオフセットからログを消費します。 重要 VVR 11.1 より前のバージョンでは、consumer_group 値はサポートされていません。指定されたコンシューマグループによって記録されたオフセットからログを消費するには、consumeFromCheckpoint を true に設定します。この場合、この起動モードは適用されません。
  startTime	ログの消費を開始する時刻です。	String	いいえ	現在時刻	書式：yyyy-MM-dd hh:mm:ss。 このオプションは、startupMode が timestamp に設定されている場合にのみ有効です。 説明 startTime および stopTime オプションは、SLS の __timestamp__ フィールドではなく、__receive_time__ フィールドに基づいています。
  stopTime	ログの消費を終了する時刻です。	String	いいえ	なし	書式：yyyy-MM-dd hh:mm:ss。 説明 このオプションは、履歴ログを消費する場合にのみ使用します。過去の時刻を指定してください。将来の時刻を指定すると、新しいログが書き込まれない場合に予期せず消費が停止する可能性があります。これは、エラーメッセージのない切断されたデータストリームとして表示されます。 ログの消費が完了した後に Flink プログラムを終了するには、exitAfterFinish も true に設定してください。
  consumerGroup	コンシューマグループの名称です。	String	いいえ	なし	コンシューマグループは消費の進行状況を記録します。任意のカスタム名を指定できます。 説明 複数のジョブによる共同消費のために、コンシューマグループを共有することはできません。異なるジョブには異なるコンシューマグループを使用してください。異なるジョブで同じコンシューマグループを使用すると、各ジョブがすべてのデータを消費します。Flink が SLS からデータを消費する際には、SLS のコンシューマグループを通じてシャードが割り当てられません。そのため、同じコンシューマグループを共有していても、各ジョブが独立してすべてのメッセージを消費します。
  consumeFromCheckpoint	指定されたコンシューマグループに保存されたチェックポイントからログを消費するかどうかを指定します。	String	いいえ	false	true：このパラメーターを true に設定する場合、コンシューマグループも指定する必要があります。Flink は、コンシューマグループに格納されたチェックポイントからログを消費します。コンシューマグループにチェックポイントが存在しない場合は、startTime パラメーターで指定された時刻からログを消費します。 false（デフォルト）：Flink は、指定されたコンシューマグループに保存されたチェックポイントからログを消費しません。 重要 このオプションは VVR 11.1 以降ではサポートされていません。VVR 11.1 以降では、startupMode を consumer_group に設定してください。
  maxRetries	SLS からの読み取りが失敗した場合の再試行回数です。	String	いいえ	3	該当なし。
  batchGetSize	1 回のリクエストで読み取るロググループの数です。	String	いいえ	100	batchGetSize を 1000 未満の値に設定してください。それ以外の場合、エラーが発生します。
  exitAfterFinish	データ消費が完了した後に Flink プログラムを終了するかどうかを指定します。	String	いいえ	false	true：データ消費が完了した後に Flink プログラムが終了します。 false（デフォルト）：データ消費が完了した後も Flink プログラムは終了しません。
  query 重要 このオプションは VVR 11.3 で非推奨となりましたが、後続のバージョンでも互換性が維持されています。	SLS データを消費する前にデータを前処理するために使用するクエリ文です。	String	いいえ	デフォルト値はありません	クエリオプションを使用して、SLS データを消費する前にフィルター処理することで、すべてのデータを Flink に読み込むことを回避し、コスト削減および処理速度の向上を図ることができます。 例: 'query' = '*| where request_method = ''GET''' は、Flink がログを読み取る前に、request_method フィールドが GET に等しいログをフィルターします。 説明 クエリは、SPL 構文を使用して記述してください。 重要 このオプションは VVR 8.0.1 以降でのみサポートされます。 この機能は SLS の課金対象となります。詳細については、「課金」をご参照ください。
  processor	SLS コンシューマプロセッサーです。query および processor の両方が設定されている場合、query が優先されます。	String	いいえ	なし	processor オプションを使用して、SLS データを消費する前にフィルター処理することで、すべてのデータを Flink に読み込むことを回避し、コスト削減および処理速度の向上を図ることができます。query の代わりに processor を使用することを推奨します。 たとえば、 'processor' = 'test-filter-processor' は、SLS コンシューマプロセッサーを適用して、Flink が SLS データを読み取る前にデータをフィルター処理します。 説明 プロセッサーは、SPL 構文を使用して記述してください。SLS コンシューマプロセッサーの作成および更新の詳細については、「コンシューマプロセッサーの管理」をご参照ください。 重要 このオプションは 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 以降でのみサポートされます。

データ型マッピング

制限事項

リアルタイムコンピューティングエンジン Ververica Runtime (VVR) 11.1 以降でのみサポートされます。

構文

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

構成オプション

データ型マッピング

データインジェストのデータ型マッピングは以下のとおりです：

スキーマ推論および進化

• 事前消費およびスキーマ初期化

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

  説明

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

• プライマリキー

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

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

• スキーマ推論および進化

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

  • 推論されたスキーマに現在のスキーマにないフィールドが含まれている場合、それらのフィールドを現在のスキーマに追加し、nullable 列追加イベントを生成します。

  • 推論されたスキーマに現在のスキーマに存在するフィールドが含まれていない場合、それらのフィールドを保持し、値を NULL に設定します。列削除イベントは生成しません。

  SLS コネクタは、すべてのフィールドを文字列として推論します。現在のところ、列の追加のみがサポートされています。新しい列は現在のスキーマの末尾に追加され、nullable としてマークされます。