Realtime Compute for Apache Flink:Kafka connector

パラメーター

• 一般

  パラメーター	説明	必須	タイプ	デフォルト	備考
  type	ソースまたはシンクのタイプです。	はい	文字列	–	値は kafka である必要があります。
  name	ソースまたはシンクの名前です。	いいえ	文字列	–	–
  properties.bootstrap.servers	Kafka ブローカーのアドレスです。	はい	文字列	–	形式は host:port,host:port,host:port で、コンマ (,) で区切ります。
  properties.*	Kafka クライアントの設定プロパティです。	いいえ	文字列	–	プロパティキーは、Apache Kafka 公式ドキュメントの「プロデューサー設定」および「コンシューマー設定」で定義されている有効なオプションである必要があります。 Realtime Compute for Apache Flink は properties. プレフィックスを削除してから、残りのキーと値のペアを基になる Kafka クライアントに渡します。例えば、'properties.allow.auto.create.topics' = 'false' を設定すると、トピックの自動作成が無効になります。
  key.format	Kafka メッセージキーのシリアル化およびデシリアル化フォーマットです。	いいえ	文字列	–	ソースの場合、json フォーマットのみサポートされます。 シンクの場合、有効な値は次のとおりです。 csv json 説明 このオプションは、Realtime Compute for Apache Flink 11.0.0 以降でのみサポートされます。
  value.format	Kafka メッセージ値のシリアル化およびデシリアル化フォーマットです。	いいえ	文字列	debezium-json	ソースの場合、有効な値は次のとおりです。 debezium-json canal-json json シンクの場合、有効な値は次のとおりです。 debezium-json canal-json canal-protobuf 説明 debezium-json および canal-json フォーマットは、Realtime Compute for Apache Flink バージョン 8.0.10 以降でサポートされます。 json フォーマットは、Realtime Compute for Apache Flink バージョン 11.0.0 以降でサポートされます。

• ソース パラメーター

  パラメータ	説明	必須	型	デフォルト	備考
  topic	読み込むトピック。	いいえ	文字列	–	複数のトピックをサブスクライブするには、名前をセミコロン（;）で区切ります。例： topic-1;topic-2 説明 このパラメータまたは topic-pattern のいずれかを指定する必要がありますが、両方を指定することはできません。
  topic-pattern	サブスクライブするトピックの名前に一致する正規表現。	いいえ	文字列	–	例： user_event_.* ： user_event_ で始まるすべてのトピックに一致します。 prod\.logs\..* ： prod.logs. で始まるトピックに一致します（. はエスケープする必要があります）。 説明 このパラメータまたは topic のいずれかを指定する必要がありますが、両方を指定することはできません。
  properties.group.id	コンシューマーグループ ID。	いいえ	文字列	–	新しいコンシューマーグループ ID を指定する場合、properties.auto.offset.reset パラメータを earliest または latest に設定して、初期の開始オフセットを指定する必要があります。
  scan.startup.mode	Kafka コンシューマーの開始オフセット。	いいえ	文字列	group-offsets	有効な値： earliest-offset ：最も古い利用可能なオフセットから読み取りを開始します。 latest-offset ：最新のオフセットから読み取りを開始します。 group-offsets（デフォルト値） ： properties.group.id で指定されたコンシューマーグループのコミット済みオフセットから読み取りを開始します。 timestamp ： scan.startup.timestamp-millis で指定されたタイムスタンプから読み取りを開始します。 specific-offsets ： scan.startup.specific-offsets で指定されたオフセットから読み取りを開始します。 説明 このパラメータは、ジョブがステートレスで起動する場合にのみ適用されます。ステートフル ジョブが起動すると、常にその状態に保存されているオフセットから消費します。
  scan.startup.specific-offsets	scan.startup.mode に specific-offsets を設定した場合のパーティションごとの開始オフセット。	いいえ	文字列	–	例： partition:0,offset:42;partition:1,offset:300
  scan.startup.timestamp-millis	scan.startup.mode に timestamp を設定した場合のミリ秒単位の開始タイムスタンプ。	いいえ	Long	–	単位はミリ秒です。
  scan.topic-partition-discovery.interval	トピック内の新しいパーティションを動的に検出する間隔。	いいえ	期間	5min	コネクタは定期的に新しいパーティションを検出し、そこから読み取ります。topic-pattern を使用する場合、コネクタはパターンに一致する新しいトピックも検出します。検出を無効にするには、この値を 0 以下に設定します。
  scan.check.duplicated.group.id	properties.group.id で指定されたコンシューマーグループの重複をチェックするかどうかを指定します。	いいえ	Boolean	false	有効な値： true ：ジョブが開始する前に重複するコンシューマーグループをチェックします。重複が見つかった場合、ジョブは失敗します。 false ：重複をチェックせずにジョブを開始します。
  schema.inference.strategy	スキーマ推論戦略。	いいえ	文字列	continuous	有効な値： continuous ：各データレコードのスキーマを解析します。スキーマに互換性がない場合、システムはより広いスキーマを推論し、スキーマ変更イベントを生成します。 static ：ジョブの開始時にスキーマ解析を 1 回だけ実行します。その後、データはこの初期スキーマに基づいて解析され、スキーマ変更イベントは生成されません。 説明 スキーマ解析の詳細については、「スキーマ解析と進化のポリシー」をご参照ください。 この設定オプションは、Ververica Runtime (VVR) 8.0.11 以降でのみサポートされています。
  scan.max.pre.fetch.records	初期スキーマ推論のためにパーティションごとに消費するメッセージの最大数。	いいえ	整数	50	データ処理が始まる前に、システムは各パーティションから指定された数の最新メッセージをプリフェッチして消費し、スキーマを初期化します。
  key.fields-prefix	名前の競合を避けるため、メッセージキーのフィールド名に付与するプレフィックス。	いいえ	文字列	–	たとえば、このパラメータが key_ に設定され、メッセージキーに a という名前のフィールドが含まれている場合、解析されたフィールド名は key_a になります。 説明 key.fields-prefix の値は、value.fields-prefix の値のプレフィックスにすることはできません。
  value.fields-prefix	名前の競合を避けるため、メッセージ値のフィールド名に付与するプレフィックス。	いいえ	文字列	–	たとえば、このパラメータが value_ に設定され、メッセージ値に b という名前のフィールドが含まれている場合、解析されたフィールド名は value_b になります。 説明 value.fields-prefix の値は、key.fields-prefix の値のプレフィックスにすることはできません。
  metadata.list	ダウンストリームのシンクに渡すメタデータ列。	いいえ	文字列	–	利用可能なメタデータ列には、topic、partition、offset、timestamp、timestamp-type、headers、および leader-epoch があります。列名はコンマで区切ります。
  scan.value.initial-schemas.ddls	特定のテーブルの初期スキーマを定義する DDL ステートメント。	いいえ	文字列	–	セミコロン（;）を使用して複数の DDL ステートメントを区切ります。たとえば、CREATE TABLE db1.t1 (id BIGINT, name VARCHAR(10)); CREATE TABLE db1.t2 (id BIGINT); を使用して、テーブル db1.t1 と db1.t2 の初期スキーマをそれぞれ指定します。 DDL で定義されたテーブルスキーマは、シンク先のテーブルと一致し、Flink SQL の構文に準拠している必要があります。 説明 この設定オプションは、Ververica Runtime (VVR) 11.5 以降でのみサポートされています。
  ingestion.ignore-errors	データ解析エラーを無視するかどうかを指定します。	いいえ	Boolean	false	説明 この設定オプションは、Ververica Runtime (VVR) 11.5 以降でのみサポートされています。
  ingestion.error-tolerance.max-count	ジョブが失敗する前に許容される解析エラーの最大数。ingestion.ignore-errors が true に設定されている場合のみ有効です。	いいえ	整数	-1	このパラメータは、ingestion.ignore-errors が true に設定されている場合にのみ適用されます。値 -1 はエラーを無制限に許容することを示し、解析例外によってジョブが失敗することはありません。 説明 この設定オプションは、Ververica Runtime (VVR) 11.5 以降でのみサポートされています。
  scan.duplicate-field.strategy	キーと値から解析された、重複するフィールド名の処理方法を指定します。	いいえ	文字列	EXCEPTION	有効な値： EXCEPTION ：キーと値に重複するフィールドが存在する場合に例外をスローします。これは VVR 11.6 以前のデフォルトの動作です。 PREFER_KEY ：フィールドが重複する場合、キー側のフィールド値を使用します。 PREFER_VALUE ：フィールドが重複する場合、値側のフィールド値を使用します。 説明 この設定オプションは、Ververica Runtime (VVR) 11.7 以降でのみサポートされています。

  • Debezium JSON フォーマットのパラメーター

    パラメーター	必須	タイプ	デフォルト	説明
    debezium-json.distributed-tables	不要	Boolean	false	単一の Debezium JSON テーブルのデータが複数のパーティションに分散されている場合は、true に設定します。 説明 この設定オプションは、Ververica Runtime (VVR) 8.0.11 以降でのみサポートされています。 重要 このパラメーターを変更するには、ステートレス起動が必要です。
    debezium-json.schema-include	不要	Boolean	false	Debezium JSON メッセージにスキーマを含めます。これは、Debezium Kafka Connect 設定の value.converter.schemas.enable プロパティに対応します。 有効な値： true：Debezium JSON メッセージにスキーマが含まれます。 false：Debezium JSON メッセージにスキーマが含まれません。
    debezium-json.ignore-parse-errors	不要	Boolean	false	有効な値： true：解析例外が発生する行をスキップします。 false：エラーがスローされてジョブが失敗します。
    debezium-json.infer-schema.primitive-as-string	不要	Boolean	false	テーブルスキーマを解析する際に、すべてのプリミティブ型を String として解析します。 有効な値： true：すべてのプリミティブ型を String として解析します。 false：デフォルトルールに基づいて型を解析します。

  • Canal JSON フォーマットのパラメーター

    パラメーター	必須	タイプ	デフォルト	説明
    canal-json.distributed-tables	不要	Boolean	false	Canal JSON の単一テーブルのデータが複数のパーティションに分散されている場合は、このオプションを有効にする必要があります。 説明 この設定オプションは、VVR 8.0.11 以降でのみサポートされています。 重要 このパラメーターを変更するには、ステートレス起動が必要です。
    canal-json.database.include	不要	String	–	Canal レコード内の database メタデータフィールドで変更ログをフィルタリングするための、オプションの正規表現です。一致するデータベースのレコードのみが処理されます。この正規表現は、Java の Pattern クラスと互換性があります。
    canal-json.table.include	不要	String	–	Canal レコード内の table メタデータフィールドで変更ログをフィルタリングするための、オプションの正規表現です。一致するテーブルのレコードのみが処理されます。この正規表現は、Java の Pattern クラスと互換性があります。
    canal-json.ignore-parse-errors	不要	Boolean	false	有効な値： true：解析例外が発生した場合、現在の行をスキップします。 false：エラーがスローされ、ジョブの起動に失敗します。
    canal-json.infer-schema.primitive-as-string	不要	Boolean	false	テーブルスキーマを解析する際に、すべてのプリミティブ型を String として解析します。 有効な値： true：すべてのプリミティブ型を String として解析します。 false：デフォルトルールに基づいて型を解析します。
    canal-json.infer-schema.strategy	不要	String	AUTO	テーブルスキーマの解析ストラテジー。 有効な値： AUTO：JSON データからスキーマを自動的に解析します。解析の失敗を防ぐため、データに sqlType フィールドが含まれていない場合に推奨されます。 SQL_TYPE：Canal JSON データ内の sqlType 配列からスキーマを解析します。データに sqlType フィールドが含まれている場合、より正確な型を取得するために SQL_TYPE を設定することを推奨します。 MYSQL_TYPE：Canal JSON データ内の mysqlType 配列からスキーマを解析します。 sqlType の型マッピングルールの詳細については、「Canal JSON's Schema parse」をご参照ください。 説明 この設定オプションは、VVR 11.1 以降でのみサポートされています。 MYSQL_TYPE の値は、VVR 11.3 以降でサポートされています。
    canal-json.mysql.treat-mysql-timestamp-as-datetime-enabled	不要	Boolean	true	MySQL の TIMESTAMP 型を CDC の TIMESTAMP 型にマッピングします。 true：MySQL の TIMESTAMP 型は、CDC の TIMESTAMP 型にマッピングされます。 false：MySQL の TIMESTAMP 型は、CDC の TIMESTAMP_LTZ 型にマッピングされます。
    canal-json.mysql.treat-tinyint1-as-boolean.enabled	不要	Boolean	true	MYSQL_TYPE 解析ストラテジーを使用する場合に、MySQL の TINYINT(1) 型を CDC の BOOLEAN 型にマッピングするかどうかを制御します。 true：MySQL の TINYINT(1) 型は、CDC の BOOLEAN 型にマッピングされます。 false：MySQL の TINYINT(1) 型は、CDC の TINYINT(1) 型にマッピングされます。 このオプションは、canal-json.infer-schema.strategy が MYSQL_TYPE に設定されている場合にのみ適用されます。

  • JSON フォーマットのパラメーター

    パラメーター	必須	タイプ	デフォルト	説明
    json.timestamp-format.standard	不要	String	SQL	入力データと出力データのタイムスタンプフォーマット。 SQL：yyyy-MM-dd HH:mm:ss.s{precision} フォーマット (例：2020-12-30 12:13:14.123) の入力タイムスタンプを解析します。 ISO-8601：yyyy-MM-ddTHH:mm:ss.s{precision} フォーマット (例：2020-12-30T12:13:14.123) の入力タイムスタンプを解析します。
    json.ignore-parse-errors	不要	Boolean	false	有効な値： true：解析例外が発生した場合、現在の行をスキップします。 false：エラーがスローされ、ジョブの起動に失敗します。
    json.infer-schema.primitive-as-string	不要	Boolean	false	テーブルスキーマを解析する際に、すべてのプリミティブ型を String として解析します。 有効な値： true：すべてのプリミティブ型を String として解析します。 false：デフォルトルールに基づいて型を解析します。
    json.infer-schema.flatten-nested-columns.enable	不要	Boolean	false	JSON データ内のネストされた列を再帰的に展開します。有効な値： true：ネストされた列を再帰的に展開します。 false：ネストされた列を String として扱います。
    json.decode.parser-table-id.fields	不要	String	–	JSON 形式のデータを解析する際に、指定された JSON フィールドの値を使用して tableId を生成します。複数のフィールドの値は、英語のコンマ , で連結されます。たとえば、JSON データが {"col0":"a", "col1","b", "col2","c"} の場合、生成される結果は次のようになります。 設定 tableId col0 a col0,col1 a.b col0,col1,col2 a.b.c
    json.infer-schema.fixed-types	不要	String	–	JSON データをパースする際、特定のフィールドのデータ型を指定できます。複数のフィールドを区切るには、カンマ , を使用します。たとえば、id BIGINT, name VARCHAR(10) は、id フィールドが BIGINT 型、name フィールドが VARCHAR(10) 型であることを指定します。 説明 この設定オプションは、VVR 11.5 以降でのみサポートされています。 VVR バージョン 11.5 でこの設定を使用する場合は、設定 scan.max.pre.fetch.records: 0 も追加する必要があります。
    json.decode.empty-value-as-delete.enabled	不要	Boolean	false	Kafka の圧縮トピック内の (値が空の) tombstone メッセージを DELETE イベントとして解析するかどうかを指定します。圧縮トピックのミラーリングや CDC 削除シグナルなど、空の値が削除セマンティクスを表すシナリオで使用します。 説明 この設定オプションは、VVR 11.7 以降でのみサポートされています。

• シンクテーブルのパラメーター

  パラメーター	説明	必須	タイプ	デフォルト	備考
  type	シンクのタイプ。	はい	String	–	値は kafka である必要があります。
  name	シンクの名前。	いいえ	String	–	–
  topic	Kafka トピック名。	いいえ	String	–	このパラメーターが指定されている場合、すべてのデータがこのトピックに書き込まれます。 説明 このパラメーターが指定されていない場合、各レコードは TableID にちなんで名付けられたトピックに書き込まれます。TableID は、データベース名とテーブル名をピリオド (.) で結合して作成されます (例： databaseName.tableName)。
  partition.strategy	Kafka パーティションへの書き込み戦略。	いいえ	String	all-to-zero	有効な値： all-to-zero (デフォルト)： すべてのデータをパーティション 0 に書き込みます。 hash-by-key： プライマリキーのハッシュ値に基づいてデータをパーティションに書き込みます。これにより、同じプライマリキーを持つレコードを同じパーティションに書き込み、その順序を保持します。
  sink.tableId-to-topic.mapping	アップストリームテーブル名からダウンストリームの Kafka トピック名へのマッピング。	いいえ	String	–	マッピングをセミコロン (;) で区切ります。各マッピング内では、アップストリームテーブル名とダウンストリームの Kafka トピック名をコロン (:) で区切ります。テーブル名には正規表現を使用できます。複数のテーブルを同じトピックにマッピングするには、テーブル名をコンマ (,) で区切ります。例： mydb.mytable1:topic1;mydb.mytable2:topic2。 説明 このパラメーターを使用すると、元のテーブル名情報を保持したまま、マッピングされたトピックを変更できます。

  • Debezium JSON フォーマットのパラメーター

    パラメーター	必須	タイプ	デフォルト	説明
    debezium-json.include-schema.enabled	いいえ	ブール	false	Debezium JSON データにスキーマ情報を含めます。
    debezium-json.emit.full-table-id.enabled	いいえ	ブール	false	完全な 3 部構成のテーブル ID を Debezium JSON メタデータフィールドに書き込みます。 このパラメーターが有効な場合、マッピングは次のようになります： CDC テーブル ID の部分 Debezium JSON キー 名前空間 db スキーマ schema テーブル table このパラメーターが無効な場合、マッピングは次のようになります： CDC テーブル ID の部分 Debezium JSON キー 名前空間 マッピングなし スキーマ db テーブル table 説明 このパラメーターは、Ververica Runtime (VVR) 11.6 以降でのみサポートされています。

テーブルスキーマの初期化

テーブルスキーマには、列とデータ型、データベース名とテーブル名、およびプライマリキーが含まれます。次のセクションでは、それぞれの初期化方法について説明します。

• 列とデータ型の情報

source:
  type: kafka
  name: Kafka Source
  properties.bootstrap.servers: host:9092
  topic: test-topic
  value.format: json
  scan.startup.mode: earliest-offset
  # 初期テーブルスキーマを設定する
  scan.value.initial-schemas.ddls: CREATE TABLE db1.t1 (id BIGINT, name VARCHAR(10)); CREATE TABLE db1.t2 (id BIGINT);

source:
  type: kafka
  name: Kafka Source
  properties.bootstrap.servers: host:9092
  topic: test-topic
  value.format: json
  scan.startup.mode: earliest-offset
  # 特定のフィールドを固定型に設定する
  json.infer-schema.fixed-types: id BIGINT, name VARCHAR(10)
  scan.max.pre.fetch.records: 0

• データベースとテーブルの情報

  • Canal JSON および Debezium JSON フォーマットの場合、コネクタは各メッセージからデータベース名とテーブル名を含むテーブル情報を解析します。

  • JSON フォーマットの場合、デフォルトでは、テーブル情報にはテーブル名のみが含まれ、これはデータを含むトピックの名前です。データにデータベースとテーブルの情報が含まれている場合は、json.decode.parser-table-id.fields パラメーターを使用して、この情報を含むフィールドを指定できます。これらのフィールドは、データベース名とテーブル名にマッピングされます。以下は設定例です：

    source:
      type: kafka
      name: Kafka Source
      properties.bootstrap.servers: host:9092
      topic: test-topic
      value.format: json
      scan.startup.mode: earliest-offset
      # col1 フィールドの値をデータベース名、col2 フィールドの値をテーブル名として使用する
      json.decode.parser-table-id.fields: col1,col2

    この設定により、コネクタは、col1 フィールドの値をデータベース名、col2 フィールドの値をテーブル名として、各レコードを対応するテーブルに送信します。

• プライマリキー情報

  • Canal JSON フォーマットの場合、JSON データ内の pkNames フィールドがテーブルのプライマリキーを定義します。

  • Debezium JSON および JSON フォーマットの場合、データにはプライマリキー情報が含まれません。transform ルールを使用して、テーブルに手動でプライマリキーを追加できます：

    transform:
      - source-table: \.*.\.*
        projection: \*
        primary-keys: key1, key2

スキーマ解析とスキーマ進化

テーブルスキーマが初期化された後、schema.inference.strategy が static に設定されている場合、Kafka コネクタは初期テーブルスキーマに基づいて各メッセージのメッセージ値を解析し、スキーマ変更イベントを生成しません。schema.inference.strategy が continuous に設定されている場合、Kafka コネクタは各 Kafka メッセージのメッセージ値を解析し、その物理列を識別し、結果のスキーマを現在保持しているスキーマと比較します。スキーマに一貫性がない場合、コネクタはそれらをマージしようとし、対応するテーブルスキーマ変更イベントを生成します。マージルールは次のとおりです：

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

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

• 同じ名前の列は次のように処理されます：

  • 列のデータ型は同じで精度が異なる場合、より大きな精度の型が使用され、列型変更イベントが生成されます。

  • 列のデータ型が異なる場合、システムは以下の型階層ツリーから最小の共通親型を見つけます。システムは、その共通親型を列に使用し、列型変更イベントを生成します。

    

• サポートされているスキーマ進化ポリシー：

  • 列の追加：コネクタは新しい列をスキーマの末尾に追加し、そのデータを同期します。新しい列は null 許容に設定されます。

  • 列の削除：列削除イベントは生成されません。代わりに、その列の後続のデータは NULL で埋められます。

  • 列名の変更：コネクタはこれを古い列の削除と新しい列の追加として扱います。新しい列はスキーマの末尾に追加され、元の列の値は NULL で埋められます。

  • 列の型の変更：

    • 列の型変更をサポートする下流シンクの場合、Flink CDC ジョブは、下流シンクがそれらを処理するように設定されていれば、型変更 (例：INT から BIGINT へ) を処理できます。この機能は、特定のシンクがサポートする列型変更ルールに依存します。サポートされているルールについては、お使いのシンクのドキュメントをご参照ください。

    • Hologres のように列の型変更をサポートしない下流シンクの場合、型拡張を使用できます。この機能は、ジョブの開始時に下流シンクに、より広いデータ型を持つテーブルを作成します。列の型が変更された場合、新しい型が下流シンクで定義されたより広い型の範囲内であれば、システムはその変更を許容できます。

• サポートされていないスキーマ変更：

  • プライマリキーやインデックスなどの制約の変更。

  • 列を NULLABLE から NOT NULL に変更すること。

• Canal JSON スキーマ解析

  Canal JSON データには、データ列の正確な型情報を記録するオプションの sqlType フィールドが含まれている場合があります。より正確なスキーマを取得するには、canal-json.infer-schema.strategy を SQL_TYPE に設定して、sqlType フィールドの型を使用できます。型マッピングは次のとおりです：

  JDBC 型	型コード	CDC 型
  BIT	-7	BOOLEAN
  BOOLEAN	16
  TINYINT	-6	TINYINT
  SMALLINT	5	SMALLINT
  INTEGER	4	INT
  BIGINT	-5	BIGINT
  DECIMAL	3	DECIMAL(38,18)
  NUMERIC	2
  REAL	7	FLOAT
  FLOAT	6
  DOUBLE	8	DOUBLE
  BINARY	-2	BYTES
  VARBINARY	-3
  LONGVARBINARY	-4
  BLOB	2004
  DATE	91	DATE
  TIME	92	TIME
  TIMESTAMP	93	TIMESTAMP
  CHAR	1	STRING
  VARCHAR	12
  LONGVARCHAR	-1
  その他のデータ型