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' は、Flink が SLS からデータを読み取る前に、SLS プロセッサがデータをフィルタリングすることを示します。 説明 このオプションは Log Service (SLS) の SPL 言語を使用します。詳細については、「SPL 構文」をご参照ください。SLS プロセッサの作成または更新方法については、「プロセッサの管理」をご参照ください。 重要 このオプションは VVR 11.3 以降でのみサポートされています。 この機能には Log Service (SLS) の料金が発生します。詳細については、「Log Service の課金」をご参照ください。
  preserveRawBytes	enableNewSource が true に設定されている場合に、SLS が運ぶ生のバイトを直接フェッチして保持するかどうかを指定します。	Boolean	いいえ	false	このオプションを有効にすると、BINARY/VARBINARY フィールドは生の byte[] を直接読み取り、CHAR/VARCHAR フィールドは生のバイトから文字列データを構築します。他のフィールドは、引き続きデフォルトの変換ロジックを使用します。この動作は、古いソースの動作と一致します。 説明 このオプションは VVR 11.8 以降でのみサポートされています。 このオプションは、enableNewSource が true に設定されている場合にのみ有効です。

• sink 固有

  パラメーター	説明	タイプ	必須	デフォルト	備考
  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 許容に設定します。