Realtime Compute for Apache Flink: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}'
);

オプション付き

• 一般

  パラメーター	説明	型	必須	デフォルト	備考
  connector	使用するコネクタ。	String	はい	なし	sls に設定します。
  endPoint	Log Service (SLS) のエンドポイント。	String	はい	なし	Log Service (SLS) の VPC アクセスアドレスを指定します。詳細については、「サービスエンドポイント」をご参照ください。 説明 デフォルトでは、Realtime Compute for Apache Flink はインターネットにアクセスできません。Virtual Private Cloud (VPC) からインターネットへのアクセスを有効にするには、NAT Gateway を使用します。詳細については、「インターネットへのアクセス方法」をご参照ください。 SLS へのインターネット経由のアクセスは推奨しません。やむを得ずインターネット経由でアクセスする場合は、HTTPS を使用し、転送アクセラレーションを有効にしてください。詳細については、「転送アクセラレーションの管理」をご参照ください。
  project	SLS プロジェクトの名前。	String	はい	なし	N/A
  logStore	Logstore または Metricstore の名前。	String	はい	なし	Logstore のデータは、Metricstore のデータと同様に消費されます。
  accessId	ご利用の Alibaba Cloud アカウントの AccessKey ID。	String	はい	なし	詳細については、「AccessKey ペアの取得」をご参照ください。 重要 AccessKey ペアが公開されないようにするため、AccessKey ID および AccessKey Secret を変数で指定することを推奨します。詳細については、「プロジェクト変数」をご参照ください。
  accessKey	ご利用の Alibaba Cloud アカウントの AccessKey Secret。	String	はい	なし

• ソース固有

  パラメーター	説明	型	必須	デフォルト	備考
  enableNewSource	FLIP-27 インターフェイスを実装する新しいデータソースを使用するかどうかを指定します。	Boolean	いいえ	false	新しいソースは、シャードの変更に自動的に適応し、すべてのソースサブタスク間でシャードを可能な限り均等に分散できます。 重要 このオプションは VVR 8.0.9 以降でのみサポートされています。VVR 11.1 以降ではデフォルト値が true です。 このオプションの値を変更すると、保存された状態からジョブを復元できなくなります。履歴オフセットから消費を再開するには、まず consumerGroup オプションを使用してジョブを開始し、SLS コンシューマーグループに消費進捗を記録します。その後、consumeFromCheckpoint オプションを true に設定し、状態なしでジョブを再起動します。 Logstore に読み取り専用のシャードがある場合、一部のサブタスクが自身のシャードの処理を終了した後も、他のシャードからデータを要求し続ける可能性があります。これによりワークロードが不均衡になり、パフォーマンスに影響を与えることがあります。この問題を軽減するには、並列度を調整するか、スケジューリング戦略を最適化するか、小さなシャードをマージしてシャード数を減らし、タスク割り当てを簡素化してください。
  shardDiscoveryIntervalMs	動的なシャード検出の間隔。	Long	いいえ	60000	動的検出を無効にするには、このオプションを負の値に設定します。単位：ミリ秒。 説明 値は 60,000 ミリ秒 (1 分) 以上である必要があります。 このオプションは 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 ジョブでは、異なるコンシューマーグループを使用する必要があります。複数の Flink ジョブで同じコンシューマーグループを使用すると、それらは連携せず、各ジョブがすべてのデータを消費します。これは、Flink が SLS からデータを消費する際に、SLS コンシューマーグループを使用してパーティション割り当てを行わないためです。その結果、同じコンシューマーグループを共有していても、各コンシューマーは独立してメッセージを消費します。
  consumeFromCheckpoint	コンシューマーグループのチェックポイントから消費するかどうかを指定します。	String	いいえ	false	true：コンシューマーグループも指定する必要があります。Flink プログラムは、コンシューマーグループに保存されたチェックポイントからログの消費を開始します。コンシューマーグループに対応するチェックポイントがない場合、startTime 設定値から消費を開始します。 false (デフォルト値)：指定されたコンシューマーグループに保存されたチェックポイントからログの消費を開始しません。 重要 このパラメーターは VVR 11.1 以降ではサポートされていません。これらのバージョンでは、startupMode オプションを consumer_group に設定する必要があります。
  maxRetries	SLS からの読み取り失敗後の再試行回数。	String	いいえ	3	N/A
  batchGetSize	1 回のリクエストで読み取るロググループの数。	String	いいえ	100	batchGetSize 設定は 1000 を超えることはできません。超えるとエラーが報告されます。
  exitAfterFinish	すべてのデータを消費した後に Flink ジョブを終了するかどうかを指定します。	String	いいえ	false	true：Flink プログラムはすべてのデータを消費した後に終了します。 false (デフォルト)：Flink プログラムはデータ消費完了後も終了しません。
  query 重要 このオプションは VVR 11.3 で非推奨となりましたが、以降のバージョンでも互換性があります。	消費前にデータを前処理するためのクエリ文。	String	いいえ	なし	このオプションを使用して、Flink が消費する前に SLS データをフィルター処理し、コストを削減し処理速度を向上させます。 たとえば、 'query' = '*| where request_method = ''GET''' は、Flink が SLS からデータを読み取る前に、request_method フィールドの値が 'GET' であるデータをマッチングすることを示します。 説明 このオプションは Log Service (SLS) の SPL 言語を使用します。詳細については、「SPL 構文」をご参照ください。 重要 このオプションは VVR 8.0.1 以降でのみサポートされています。 この機能には Log Service (SLS) 料金が発生します。詳細については、「Log Service の課金」をご参照ください。
  processor	データ前処理用の SLS プロセッサの名前。query と processor の両方が指定されている場合、query が優先され、processor は無視されます。	String	いいえ	なし	このオプションは、Flink が消費する前に SLS データをフィルター処理し、コストを削減し処理速度を向上させます。query の代わりに processor を使用することを推奨します。 たとえば、 'processor' = 'test-filter-processor' は、SLS プロセッサが Flink が SLS からデータを読み取る前にデータをフィルター処理することを示します。 説明 このオプションは Log Service (SLS) の SPL 言語を使用します。詳細については、「SPL 構文」をご参照ください。SLS プロセッサの作成または更新方法については、「プロセッサの管理」をご参照ください。 重要 このオプションは VVR 11.3 以降でのみサポートされています。 この機能には Log Service (SLS) 料金が発生します。詳細については、「Log Service の課金」をご参照ください。

• シンク固有

  パラメーター	説明	型	必須	デフォルト	備考
  topicField	ログトピックを示す __topic__ 属性を上書きするフィールドを指定します。	String	いいえ	なし	このオプションの値は、テーブル内に存在するフィールドである必要があります。
  timeField	ログ書き込み時刻を示す __timestamp__ 属性を上書きするフィールドを指定します。	String	いいえ	現在時刻	このオプションの値は、テーブル内に存在する INT フィールドである必要があります。このオプションを指定しない場合、現在時刻が使用されます。
  sourceField	ログソース (ログを生成したマシンの IP アドレスなど) を示す __source__ 属性を上書きするフィールドを指定します。	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 以降でのみサポートされています。

データ型マッピング

制限事項

この機能は Realtime Compute for Apache Flink バージョン 11.1 以降でのみサポートされています。

構文

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

パラメーター

既存のカタログの再利用

Realtime Compute for Apache Flink バージョン 11.5 以降では、Flink CDC データインジェストジョブで Data Management ページに作成された組み込み SLS カタログを参照できます。これにより、接続プロパティを手動で指定する必要がなくなります。

source:
  type: sls
  using.built-in-catalog: sls_catalog

現在、データインジェストジョブは組み込み SLS カタログから次のパラメーターを自動的に再利用できます。

• endpoint

• project

• accessId

• accessKey

これらの自動再利用パラメーターを上書きするには、YAML 構成で明示的に定義できます。YAML ファイルで定義されたパラメーターが優先されます。

データ型マッピング

fixed-types が構成されていない場合、次のデータ型マッピングが適用されます。

fixed-types が構成されている場合、システムは指定された型を使用してデータを解析します。

スキーマ推論と進化

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

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

  説明

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

• プライマリキー情報

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

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

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

  スキーマが初期化された後、schema.inference.strategy が static に設定されている場合、SLS コネクタは各ログエントリを初期スキーマに基づいて解析し、スキーマ変更イベントを生成しません。schema.inference.strategy が continuous に設定されている場合、コネクタは各ログエントリを解析し、物理カラムを推論して現在のスキーマと比較します。推論されたスキーマが現在のスキーマと不一致の場合、次のルールに従ってスキーマをマージします。

  • 推論された物理カラムに現在のスキーマにないフィールドが含まれている場合、コネクタはこれらのフィールドをスキーマに追加し、NULL 可能なカラムを追加するイベントを生成します。

  • 推論された物理カラムに現在のスキーマに存在するフィールドが不足している場合、コネクタはこれらのフィールドを保持し、データを NULL で埋め、カラム削除イベントは生成しません。

  SLS コネクタは、各ログエントリ内のすべてのフィールドのデータ型を String として推論します。現在、新しいカラムの追加のみがサポートされています。コネクタは新しいカラムをスキーマの末尾に追加し、NULL 可能に設定します。