Realtime Compute for Apache Flink:Kafka connector

概要

前提条件

注意事項

Apache Flink および Apache Kafka の既知の設計上の制限のため、トランザクション書き込みは推奨されません。sink.delivery-guarantee = 'exactly-once' を設定すると、Kafka コネクタはトランザクション書き込みを有効にしますが、次のような既知の問題があります。

• 各チェックポイントは新しいトランザクション ID を生成します。チェックポイントの間隔が短すぎると、トランザクション ID が大量に発生し、Kafka クラスターのコーディネーターがメモリ不足になり、クラスターの安定性が損なわれる可能性があります。

• 各トランザクションは新しいプロデューサーインスタンスを作成します。あまりにも多くのトランザクションが同時にコミットされると、TaskManager がメモリ不足になり、Apache Flink ジョブが不安定になる可能性があります。

• 複数の Apache Flink ジョブが同じ sink.transactional-id-prefix を使用すると、生成されたトランザクション ID が競合する可能性があります。1つのジョブで書き込み操作が失敗すると、Apache Kafka パーティションのログ開始オフセット (LSO) が進まなくなる可能性があります。これは、そのパーティションのすべてのコンシューマーに影響します。

1 回限りのセマンティクスが必要な場合は、Upsert Kafka コネクタを使用してプライマリキーテーブルに書き込み、べき等性を確保します。トランザクション書き込みを使用する必要がある場合は、「1 回限りのセマンティクスの使用上の注意」をご参照ください。

ネットワーク接続のトラブルシューティング

Realtime Compute for Apache Flink ジョブが起動に失敗したときの Timed out waiting for a node assignment エラーは、通常、Realtime Compute for Apache Flink と Kafka クラスター間のネットワーク接続の問題を示します。

SQL

SQL ジョブで Kafka コネクタをソーステーブルまたは結果テーブルとして使用します。

CREATE TABLE KafkaTable (
  `user_id` BIGINT,
  `item_id` BIGINT,
  `behavior` STRING,
  `ts` TIMESTAMP_LTZ(3) METADATA FROM 'timestamp' VIRTUAL
) WITH (
  'connector' = 'kafka',
  'topic' = 'user_behavior',
  'properties.bootstrap.servers' = 'localhost:9092',
  'properties.group.id' = 'testGroup',
  'scan.startup.mode' = 'earliest-offset',
  'format' = 'csv'
)

メタデータ列

CREATE TABLE kafka_source (
  -- メッセージトピックを `record_topic` 列として読み取る
  `record_topic` STRING NOT NULL METADATA FROM 'topic' VIRTUAL,
  -- ConsumerRecord からタイムスタンプを `ts` 列として読み取る
  `ts` TIMESTAMP_LTZ(3) METADATA FROM 'timestamp' VIRTUAL,
  -- メッセージオフセットを `record_offset` 列として読み取る
  `record_offset` BIGINT NOT NULL METADATA FROM 'offset' VIRTUAL,
  ...
) WITH (
  'connector' = 'kafka',
  ...
);

CREATE TABLE kafka_sink (
  -- `ts` 列のタイムスタンプを ProducerRecord のタイムスタンプとして Kafka に書き込む
  `ts` TIMESTAMP_LTZ(3) METADATA FROM 'timestamp' VIRTUAL,
  ...
) WITH (
  'connector' = 'kafka',
  ...
);

コネクタオプション

• 一般

  オプション	説明	型	必須	デフォルト	備考
  connector	コネクタのタイプ。	String	はい	–	値は kafka である必要があります。
  properties.bootstrap.servers	Kafka ブローカーアドレスのリスト。	String	はい	–	フォーマット：host1:port1,host2:port2,...。アドレスはカンマ (,) で区切ります。
  properties.*	Kafka クライアントの追加プロパティ。	String	いいえ	–	プロパティキーは、公式の Apache Kafka ドキュメントのプロデューサー設定およびコンシューマー設定で定義されている有効なオプションである必要があります。 Realtime Compute for Apache Flink は properties. プレフィックスを削除し、残りのキーと値のペアを基盤となる Kafka クライアントに渡します。たとえば、'properties.allow.auto.create.topics' = 'false' を設定して、自動的なトピック作成を無効にできます。 Kafka コネクタはこれらのオプションを上書きするため、この方法で設定することはできません： key.deserializer value.deserializer
  format	Kafka メッセージの値のシリアル化および逆シリアル化のためのフォーマット。	String	いいえ	–	サポートされているフォーマット： csv json avro debezium-json canal-json maxwell-json avro-confluent raw 説明 詳細については、「フォーマットオプション」をご参照ください。
  key.format	Kafka メッセージのキーのシリアル化および逆シリアル化のためのフォーマット。	String	いいえ	–	サポートされているフォーマット： csv json avro debezium-json canal-json maxwell-json avro-confluent raw 説明 この構成を使用する場合、key.options が必須です。
  key.fields	Kafka メッセージキーとして使用するテーブルスキーマのフィールド。	String	いいえ	–	複数のフィールド名はセミコロン (;) で区切ります。例：'field1;field2'。
  key.fields-prefix	値フィールドとの名前の競合を防ぐための、すべてのキーフィールドのカスタムプレフィックス。	String	いいえ	–	このプレフィックスは、キーフィールドと値フィールドを区別するために使用されます。キーをシリアル化する前、または逆シリアル化した後に削除されます。 説明 このオプションを使用する場合、value.fields-include は EXCEPT_KEY に設定する必要があります。
  value.format	Kafka メッセージの値のシリアル化および逆シリアル化のためのフォーマット。	String	いいえ	–	この構成は format と同等です。format または value.format のいずれか 1 つのみを設定できます。両方が設定されている場合、value.format が format をオーバーライドします。
  value.fields-include	キーフィールドが値フォーマットに含まれるかどうかを定義します。	String	いいえ	ALL	有効な値： ALL: Kafka メッセージの値に、テーブルのすべての列が含まれます。 EXCEPT_KEY：Kafka メッセージの値には、key.fields で定義されたものを除くすべてのテーブル列が含まれます。

• ソーステーブル

  オプション	説明	型	必須	デフォルト	備考
  topic	読み取るトピック。	String	いいえ	–	複数のトピックをサブスクライブするには、トピック名をセミコロン (;) で区切ります。例：'topic-1;topic-2'。 説明 このオプションまたは topic-pattern のいずれかを指定できますが、両方を指定することはできません。
  topic-pattern	サブスクライブするトピックに一致する正規表現。コンシューマーは、このパターンに一致する名前のすべてのトピックをサブスクライブします。	String	いいえ	–	例： user_event_.*: プレフィックス user_event_ を持つすべてのトピックに一致します。 prod\.logs\..*：prod.logs. で始まるトピックに一致します（. 文字はエスケープする必要があります）。 説明 このオプションまたはトピックのいずれかを指定できますが、両方を指定することはできません。
  properties.group.id	Kafka ソースコンシューマーグループ ID。	String	いいえ	KafkaSource-{Source-Table-Name}	初めてコンシューマーグループ ID を使用する場合は、properties.auto.offset.reset を earliest または latest のいずれかに設定して、初期起動オフセットを定義する必要もあります。
  scan.startup.mode	Kafka コンシューマーの開始オフセット。	String	いいえ	group-offsets	有効な値： earliest-offset: 利用可能な最も早いオフセットから読み取りを開始します。 latest-offset: 最大オフセットから読み取りを開始します。 group-offsets: 指定された properties.group.id のコミットされたオフセットから読み取りを開始します。 タイムスタンプ: 指定された scan.startup.timestamp-millis から読み取りを開始します。 specific-offsets: scan.startup.specific-offsets で指定されたオフセットから読み取りを開始します。 説明 このオプションは、ジョブがステートなしで開始される場合にのみ適用されます。ジョブがチェックポイントから再開される場合、チェックポイントステートに保存されているオフセットから読み取ります。
  scan.startup.specific-offsets	scan.startup.mode が specific-offsets の場合のパーティションごとの開始オフセット。	String	いいえ	–	例：partition:0,offset:42;partition:1,offset:300
  scan.startup.timestamp-millis	scan.startup.mode が timestamp に設定されている場合のミリ秒単位の開始タイムスタンプ。	Long	いいえ	–	単位はミリ秒です。
  scan.topic-partition-discovery.interval	パーティション検出間隔。	Duration	いいえ	5 分	コネクタは定期的に新しいパーティションを検出し、そこから読み取ります。topic-pattern を使用する場合、コネクタはパターンに一致する新しいトピックも検出します。この機能を無効にするには、間隔を非正の値に設定します。 説明 Ververica Runtime (VVR) 6.0.x では、動的パーティション検出はデフォルトで無効になっています。VVR 8.0 以降、この機能はデフォルトで有効になり、検出間隔は 5 分です。
  scan.header-filter	Kafka メッセージヘッダーに基づいてメッセージをフィルタリングします。	String	いいえ	–	ヘッダーキーとその値はコロン (:) で区切られます。複数のヘッダー条件は、論理演算子 (& および |) を使用して接続されます。NOT 論理演算子 (!) もサポートされています。たとえば、ヘッダーに depart=toy または depart=book が含まれ、かつ env=test が含まれていない場合、depart:toy|depart:book&!env:test は Kafka データを保持します。 説明 このオプションは、Ververica Runtime (VVR) 8.0.6 以降でのみサポートされます。 式中の括弧はサポートされていません。 論理演算は左から右に評価されます。 ヘッダー値は比較のために UTF-8 文字列に変換されます。
  scan.check.duplicated.group.id	他のアクティブなコンシューマーが properties.group.id をすでに使用しているかどうかを確認します。	Boolean	いいえ	false	有効な値： true：ジョブを開始する前に、重複するコンシューマーグループをチェックします。見つかった場合、競合を防ぐためにジョブは失敗します。 false：競合をチェックせずにジョブを開始します。 説明 このオプションは、Ververica Runtime (VVR) 6.0.4 以降でのみサポートされます。

• シンクテーブル

  オプション	説明	型	必須	デフォルト	備考
  topic	ターゲットトピック。	String	はい	–	–
  sink.partitioner	並列シンクインスタンスからのレコードを Kafka パーティションにマッピングします。	String	いいえ	default	有効な値： default: デフォルトの Kafka partitioner を使用します。 fixed: 各並列 sink インスタンスは、固定の Kafka パーティションに書き込みます。 round-robin: レコードはラウンドロビン方式でパーティションに分散されます。 カスタム partitioner: カスタム partitioner を使用するには、FlinkKafkaPartitioner サブクラスの完全修飾クラス名を指定します。たとえば、org.mycompany.MyPartitioner などです。
  sink.delivery-guarantee	シンクの配信保証。	String	いいえ	at-least-once	有効な値： none：保証はありません。レコードが失われたり、重複したりする可能性があります。 at-least-once: レコードが失われないことを保証しますが、重複する可能性があります。 exactly-once: Kafka トランザクションを使用して 1 回限りのセマンティクスを提供し、レコードが失われたり重複したりしないことを保証します。 説明 exactly-once セマンティクスを使用する場合、sink.transactional-id-prefix も指定する必要があります。
  sink.transactional-id-prefix	トランザクション ID プレフィックス。 sink.delivery-guarantee が exactly-once の場合に必須です。	String	はい、sink.delivery-guarantee が exactly-once の場合です。	–	sink.delivery-guarantee が exactly-once にセットされている場合にのみ必須です。
  sink.parallelism	シンク演算子の並列度。	Integer	いいえ	–	デフォルトでは、フレームワークは上流の演算子に基づいて並列度を決定します。

セキュリティと認証

CREATE TABLE KafkaTable (
  `user_id` BIGINT,
  `item_id` BIGINT,
  `behavior` STRING,
  `ts` TIMESTAMP_LTZ(3) METADATA FROM 'timestamp'
) WITH (
  'connector' = 'kafka',
  ...
  'properties.security.protocol' = 'SASL_PLAINTEXT',
  'properties.sasl.mechanism' = 'PLAIN',
  'properties.sasl.jaas.config' = 'org.apache.flink.kafka.shaded.org.apache.kafka.common.security.plain.PlainLoginModule required username="username" password="password";'
)

CREATE TABLE KafkaTable (
  `user_id` BIGINT,
  `item_id` BIGINT,
  `behavior` STRING,
  `ts` TIMESTAMP_LTZ(3) METADATA FROM 'timestamp'
) WITH (
  'connector' = 'kafka',
  ...
  'properties.security.protocol' = 'SASL_SSL',
  /* SSL 構成 */
  /* サーバーの CA 証明書のトラストストアへのパス。 */
  /* アーティファクトを使用してアップロードされたファイルは /flink/usrlib/ ディレクトリに保存されます。 */
  'properties.ssl.truststore.location' = '/flink/usrlib/kafka.client.truststore.jks',
  'properties.ssl.truststore.password' = 'test1234',
  /* クライアント認証が必要な場合は、キーストア (秘密鍵) へのパスも構成する必要があります。 */
  'properties.ssl.keystore.location' = '/flink/usrlib/kafka.client.keystore.jks',
  'properties.ssl.keystore.password' = 'test1234',
  /* サーバーのホスト名を検証するために使用されるアルゴリズム。空の文字列はホスト名の検証を無効にします。 */
  'properties.ssl.endpoint.identification.algorithm' = '',
  /* SASL 構成 */
  /* SASL メカニズムを SCRAM-SHA-256 に設定します。 */
  'properties.sasl.mechanism' = 'SCRAM-SHA-256',
  /* JAAS を構成します。 */
  'properties.sasl.jaas.config' = 'org.apache.flink.kafka.shaded.org.apache.kafka.common.security.scram.ScramLoginModule required username="username" password="password";'
)

Realtime Compute for Apache Flink コンソールの [アーティファクト] 機能を使用して、例で説明されている CA 証明書と秘密鍵をアップロードできます。 アップロードされたファイルは /flink/usrlib ディレクトリに保存されます。 my-truststore.jks という名前の CA 証明書ファイルを使用するには、WITH 句で 'properties.ssl.truststore.location' プロパティを次の 2 つの方法のいずれかで設定できます。

• 'properties.ssl.truststore.location' = '/flink/usrlib/my-truststore.jks' を設定します。この方法は、ランタイム時に Object Storage Service (OSS) からファイルを動的にダウンロードすることを回避しますが、デバッグモードをサポートしません。

• リアルタイムコンピューティングエンジンのバージョンが VVR 11.5 以降の場合、properties.ssl.truststore.location と properties.ssl.keystore.location を絶対 OSS パスに設定できます。ファイルパス形式は oss://flink-fullymanaged-<Workspace ID>/artifacts/namespaces/<Namespace name>/<file name> です。この方法では、Flink ランタイム時に OSS ファイルを動的にダウンロードし、デバッグモードをサポートします。

ソーステーブルの開始オフセット

CREATE TEMPORARY TABLE kafka_source (
  ...
) WITH (
  'connector' = 'kafka',
  ...
  -- 最も古いオフセットから消費する。
  'scan.startup.mode' = 'earliest-offset',
  -- 最新のオフセットから消費する。
  'scan.startup.mode' = 'latest-offset',
  -- コンシューマーグループ "my-group" のコミット済みオフセットから消費する。
  'properties.group.id' = 'my-group',
  'scan.startup.mode' = 'group-offsets',
  'properties.auto.offset.reset' = 'earliest', -- "my-group" が初めて使用される場合、消費は最も古いオフセットから開始される。
  'properties.auto.offset.reset' = 'latest', -- "my-group" が初めて使用される場合、消費は最新のオフセットから開始される。
  -- 指定されたタイムスタンプ (ミリ秒) から消費する：1655395200000。
  'scan.startup.mode' = 'timestamp',
  'scan.startup.timestamp-millis' = '1655395200000',
  -- 特定のオフセットから消費する。
  'scan.startup.mode' = 'specific-offsets',
  'scan.startup.specific-offsets' = 'partition:0,offset:42;partition:1,offset:300'
);

ソースオフセットのコミット

Kafka ソーステーブルは、チェックポイントが成功した後にのみコンシューマーオフセットを Kafka クラスターにコミットするため、チェックポイントの間隔が長いとコミットされたオフセットが遅れます。ソーステーブルは、実際の読み取り進捗をチェックポイントステートに保存し、システムはこれを障害復旧に使用します。コミットされたオフセットは進捗モニターとしてのみ機能し、回復には使用されないため、コミットの失敗はデータの精度に影響しません。

カスタムシンクパーティショナー

Kafka の組み込みパーティショニング戦略が要件を満たさない場合は、FlinkKafkaPartitioner クラスを拡張してカスタムパーティショナーを実装できます。開発が完了したら、コードを JAR パッケージにコンパイルし、Realtime Compute コンソールのアーティファクト機能を使用してアップロードします。JAR パッケージがアップロードされて参照された後、WITH 句の sink.partitioner パラメーターをパーティショナーの完全修飾クラス名 (例：org.mycompany.MyPartitioner) に設定します。

Kafka、Upsert Kafka、および Kafka JSON カタログ

Kafka は追加専用のイベントストリーミングプラットフォームであり、データの更新や削除をサポートしていません。ストリーミング SQL では、標準の Kafka 結果テーブルは、上流の Change Data Capture (CDC) データや、集計や結合などの演算子のリトラクションロジックを処理できません。変更やリトラクションを含むデータを書き込む必要がある場合は、Upsert Kafka 結果テーブルを使用します。

1 つまたは複数の上流データベーステーブルから Kafka への Change Data Capture (CDC) データのバッチ同期を簡素化するために、Kafka JSON カタログを使用できます。Kafka に保存されているデータが JSON 形式の場合、Kafka JSON カタログを使用すると、スキーマと WITH パラメーターを定義するステップをスキップできます。詳細については、「Kafka JSON カタログの管理」をご参照ください。

例

CREATE TEMPORARY TABLE kafka_source (
  id INT,
  name STRING,
  age INT
) WITH (
  'connector' = 'kafka',
  'topic' = 'source',
  'properties.bootstrap.servers' = '<yourKafkaBrokers>',
  'properties.group.id' = '<yourKafkaConsumerGroupId>',
  'format' = 'csv'
);

CREATE TEMPORARY TABLE kafka_sink (
  id INT,
  name STRING,
  age INT
) WITH (
  'connector' = 'kafka',
  'topic' = 'sink',
  'properties.bootstrap.servers' = '<yourKafkaBrokers>',
  'properties.group.id' = '<yourKafkaConsumerGroupId>',
  'format' = 'csv'
);

INSERT INTO kafka_sink SELECT id, name, age FROM kafka_source;

CREATE TEMPORARY TABLE kafkaTable (
  `offset` INT NOT NULL METADATA,
  `part` BIGINT NOT NULL METADATA FROM 'partition',
  PRIMARY KEY (`part`, `offset`) NOT ENFORCED
) WITH (
  'connector' = 'kafka',
  'properties.bootstrap.servers' = '<yourKafkaBrokers>',
  'topic' = 'kafka_evolution_demo',
  'scan.startup.mode' = 'earliest-offset',
  'format' = 'json',
  'json.infer-schema.flatten-nested-columns.enable' = 'true'
    -- オプション。すべてのネストされた列をフラット化します。
);

CREATE TABLE IF NOT EXISTS hologres.kafka.`sync_kafka`
WITH (
  'connector' = 'hologres'
) AS TABLE vvp.`default`.kafkaTable;

CREATE TEMPORARY TABLE kafkaTable (
  `key_id` INT NOT NULL,
  `val_name` VARCHAR(200)
) WITH (
  'connector' = 'kafka',
  'properties.bootstrap.servers' = '<yourKafkaBrokers>',
  'topic' = 'kafka_evolution_demo',
  'scan.startup.mode' = 'earliest-offset',
  'key.format' = 'json',
  'value.format' = 'json',
  'key.fields' = 'key_id',
  'key.fields-prefix' = 'key_',
  'value.fields-prefix' = 'val_',
  'value.fields-include' = 'EXCEPT_KEY'
);

CREATE TABLE IF NOT EXISTS hologres.kafka.`sync_kafka`(
WITH (
  'connector' = 'hologres'
) AS TABLE vvp.`default`.kafkaTable;

CREATE TEMPORARY TABLE kafkaTable (
  `distinct_id` INT NOT NULL,
  `properties` STRING,
  `timestamp` TIMESTAMP_LTZ METADATA,
  `date` AS CAST(`timestamp` AS DATE)
) WITH (
  'connector' = 'kafka',
  'properties.bootstrap.servers' = '<yourKafkaBrokers>',
  'topic' = 'kafka_evolution_demo',
  'scan.startup.mode' = 'earliest-offset',
  'key.format' = 'json',
  'value.format' = 'json',
  'key.fields' = 'key_id',
  'key.fields-prefix' = 'key_'
);

CREATE TABLE IF NOT EXISTS hologres.kafka.`sync_kafka` WITH (
   'connector' = 'hologres'
) AS TABLE vvp.`default`.kafkaTable
ADD COLUMN
  `order_id` AS COALESCE(JSON_VALUE(`properties`, '$.order_id'), 'default');
--COALESCE を使用して null 値を処理します。

{
  "id": 101,
  "name": "VVP",
  "properties": {
    "owner": "Alibaba Cloud",
    "engine": "Flink"
  }
}

CREATE TEMPORARY TABLE kafka_source (
  id          VARCHAR,
  `name`      VARCHAR,
  properties  ROW<`owner` STRING, engine STRING>
) WITH (
  'connector' = 'kafka',
  'topic' = 'xxx',
  'properties.bootstrap.servers' = 'xxx',
  'scan.startup.mode' = 'earliest-offset',
  'format' = 'json'
);

DataStream API

• Kafka ソースの構築

  Kafka Sourceは、Kafka Source インスタンスを作成するためのビルダー クラスを提供します。 次のサンプル コードは、input-topic Topicの最も古いOffsetからデータを消費するKafka Sourceをビルドします。 コンシューマー グループはmy-groupで、Kafka メッセージの値は文字列として逆シリアル化されます。

  Java

  KafkaSource<String> source = KafkaSource.<String>builder()
      .setBootstrapServers(brokers)
      .setTopics("input-topic")
      .setGroupId("my-group")
      .setStartingOffsets(OffsetsInitializer.earliest())
      .setValueOnlyDeserializer(new SimpleStringSchema())
      .build();
  
  env.fromSource(source, WatermarkStrategy.noWatermarks(), "Kafka Source");

  Kafka ソースを構築するには、次のプロパティを指定する必要があります。

  パラメーター	説明
  BootstrapServers	Kafka ブローカーアドレスのリストです。このプロパティは、setBootstrapServers(String) メソッドを呼び出して設定します。
  GroupId	[コンシューマーグループ] の ID です。このプロパティは、setGroupId(String) メソッドを呼び出して設定します。
  トピックまたはパーティション	サブスクライブするトピックまたはパーティション。Kafka ソースは、トピックまたはパーティションをサブスクライブするための次の 3 つの方法をサポートしています： リスト内のトピックのすべてのパーティションをサブスクライブします。 KafkaSource.builder().setTopics("topic-a","topic-b") トピックパターン：指定された正規表現に一致する名前のトピックのすべてのパーティションをサブスクライブします。 KafkaSource.builder().setTopicPattern("topic.*") パーティションのリスト。指定されたパーティションをサブスクライブできます。 final HashSet<TopicPartition> partitionSet = new HashSet<>(Arrays.asList( new TopicPartition("topic-a", 0), // トピック "topic-a" のパーティション 0 new TopicPartition("topic-b", 5))); // トピック "topic-b" のパーティション 5 KafkaSource.builder().setPartitions(partitionSet)
  デシリアライザー	Kafka メッセージを解析するために使用されるデシリアライザー。 setDeserializer(KafkaRecordDeserializationSchema) メソッドを使用してデシリアライザーを指定します。 KafkaRecordDeserializationSchema は、Kafka ConsumerRecord を解析する方法を定義します。 Kafka メッセージの値のみを解析する必要がある場合は、次のいずれかのメソッドを使用できます。 ビルダー クラスの setValueOnlyDeserializer(DeserializationSchema) メソッドを使用します。DeserializationSchema は、Kafka メッセージ 値のバイナリデータを解析する方法を定義します。 Kafka の Deserializer インターフェイスを実装するクラスを使用します。たとえば、StringDeserializer を使用して Kafka メッセージの値を文字列に解析できます。 import org.apache.kafka.common.serialization.StringDeserializer; KafkaSource.<String>builder() .setDeserializer(KafkaRecordDeserializationSchema.valueOnly(StringDeserializer.class)); 説明 完全な ConsumerRecord を解析するには、KafkaRecordDeserializationSchema インターフェイスを実装する必要があります。

  POM

  Kafka DataStream コネクタは Maven Central リポジトリで入手できます。

  <dependency>
      <groupId>com.alibaba.ververica</groupId>
      <artifactId>ververica-connector-kafka</artifactId>
      <version>${vvr-version}</version>
  </dependency>

  Kafka DataStream コネクタを使用する際は、次のプロパティを考慮してください：

  • 開始オフセット

    Kafka ソースは、オフセットイニシャライザ (OffsetsInitializer) を使用して、その開始オフセットを指定します。組み込みのイニシャライザは次のとおりです。

    オフセット初期化子	コード
    最も古いオフセットから消費を開始します。	KafkaSource.builder().setStartingOffsets(OffsetsInitializer.earliest())
    最新のオフセットから消費を開始します。	KafkaSource.builder().setStartingOffsets(OffsetsInitializer.latest())
    指定された時刻以上のタイムスタンプを持つデータの消費を開始します。単位はミリ秒です。	KafkaSource.builder().setStartingOffsets(OffsetsInitializer.timestamp(1592323200000L))
    コンシューマーグループのコミット済みオフセットから消費を開始します。コミット済みオフセットが存在しない場合は、指定されたリセット戦略 (例：最も古いオフセット) を使用します。	KafkaSource.builder().setStartingOffsets(OffsetsInitializer.committedOffsets(OffsetResetStrategy.EARLIEST))
    消費はコンシューマーグループによってコミットされたオフセットから開始され、オフセットリセットポリシーは指定されません。	KafkaSource.builder().setStartingOffsets(OffsetsInitializer.committedOffsets())

    説明

    • 組み込みの初期化子が要件を満たさない場合は、カスタムのオフセット初期化子を実装できます。

    • オフセットイニシャライザーを指定しない場合、デフォルトは OffsetsInitializer.earliest() です。

  • ストリーミングモードとバッチモード

    Kafka ソースは、ストリーミングモードとバッチモードの両方をサポートしています。デフォルトではストリーミングモードで動作し、ジョブは失敗するかキャンセルされるまで無期限に実行されます。Kafka ソースをバッチモードで実行するように設定するには、setBounded(OffsetsInitializer) を使用して停止オフセットを指定できます。すべてのパーティションが指定された停止オフセットに達すると、Kafka ソースは終了します。

    説明

    ストリーミングモードの Kafka ソースには、通常、停止 Offset はありません。ただし、テスト目的で、setUnbounded(OffsetsInitializer) を使用して、ストリーミングモードでも停止 Offset を指定できます。停止 Offset を指定するためのメソッド名が異なることに注意してください。ストリーミングモードの場合は setUnbounded、バッチモードの場合は setBounded です。

  • 動的パーティション検出

    Flink ジョブを再起動せずにトピックのスケーリングや新しいトピックの作成を処理するために、パターンでトピックをサブスクライブする際に動的パーティション検出を有効にできます。この機能はデフォルトで無効になっており、明示的に有効にする必要があります：

    KafkaSource.builder()
        .setProperty("partition.discovery.interval.ms", "10000") // 10 秒ごとに新しいパーティションを検出します。

    重要

    動的パーティション検出機能は、Kafka クラスターのメタデータ更新メカニズムに依存します。Kafka クラスターがパーティション情報をタイムリーに更新しない場合、新しいパーティションが検出されない可能性があります。Kafka クラスターの partition.discovery.interval.ms 構成が実際のシナリオと一致していることを確認してください。

  • イベント時間とウォーターマーク

    デフォルトでは、Kafka ソースは Kafka メッセージのタイムスタンプをイベント時間として使用します。カスタムのウォーターマーク戦略を定義して、メッセージ本文からイベント時間を抽出し、下流にウォーターマークを送信できます。

    env.fromSource(kafkaSource, new CustomWatermarkStrategy(), "Kafka Source With Custom Watermark Strategy")

    カスタムウォーターマーク戦略の詳細については、「ウォーターマークの生成」をご参照ください。

    説明

    ソースサブタスクがアイドル状態の場合 (たとえば、Kafka パーティションに新しいデータがない場合や、ソースの並列度が Kafka パーティションの数より高い場合)、そのサブタスクのウォーターマークは進みません。これにより、下流のウィンドウ計算がブロックされる可能性があります。

    この問題を解決するには、次のソリューションを検討してください：

    • ソースアイドルタイムアウトの構成：table.exec.source.idle-timeout プロパティを有効にして、アイドル状態のソースを一時的にアイドルとしてマークします。これにより、下流のウォーターマークが進むことができます。

    • 適切な並列度の設定：ソースの並列度が Kafka パーティションの数を超えないようにします。

  • オフセットコミット

    チェックポイントが有効な場合、Kafka ソースはチェックポイントが完了したときに現在のコンシューマーオフセットを Kafka にコミットします。これにより、Flink チェックポイントステートが Kafka ブローカー上のコミット済みオフセットと一致することが保証されます。チェックポイントが無効な場合、Kafka ソースは Kafka コンシューマーの内部的な自動定期オフセットコミットメカニズムに依存します。この機能は、enable.auto.commit および auto.commit.interval.ms Kafka コンシューマープロパティによって制御されます。

    説明

    Kafka ソースは、フォールトトレランスと回復のためにコミットされたオフセットに依存しません。オフセットのコミットは、Kafka コンシューマーとコンシューマーグループの進捗を監視するためだけです。

  • その他のプロパティ

    前述のプロパティに加えて、setProperties(Properties) および setProperty(String, String) を使用すると、Kafka ソース とその基盤となる Kafka コンシューマーに任意の プロパティ を設定できます。 Kafka ソース は、次の特定のプロパティを提供します。

    パラメーター	説明
    client.id.prefix	Kafka コンシューマーのクライアント ID プレフィックス。
    partition.discovery.interval.ms	パーティション検出の間隔 (ミリ秒)。値に -1 を指定すると、動的パーティション検出が無効になります。 説明 [バッチモード] では、このプロパティは自動的に -1 に設定されます。
    register.consumer.metrics	Flink に Kafka コンシューマーメトリックを登録します。
    その他の Kafka コンシューマー構成	Kafka コンシューマー構成の完全なリストについては、公式の Apache Kafka ドキュメントをご参照ください。 重要 正しい操作を保証するために、Kafka DataStream コネクタは、手動で構成された次のプロパティを上書きします： key.deserializer は、常に org.apache.kafka.common.serialization.ByteArrayDeserializer に上書きされます。 value.deserializer は常に org.apache.kafka.common.serialization.ByteArrayDeserializer に上書きされます。 auto.offset.reset.strategy は、OffsetsInitializer が提供するストラテジーによって上書きされます。

    次の例は、Kafka コンシューマーを PLAIN SASL メカニズムを使用し、JAAS 構成を提供するように構成する方法を示しています。

    KafkaSource.builder()
        .setProperty("sasl.mechanism", "PLAIN")
        .setProperty("sasl.jaas.config", "org.apache.kafka.common.security.plain.PlainLoginModule required username=\"username\" password=\"password\";")

  • モニタリング

    Kafka ソースは、モニタリングと診断のために Flink のメトリックシステムを通じてメトリックを公開します。

    • メトリックスコープ

      Kafka ソースリーダーのすべてのメトリックは、演算子のメトリックグループのサブグループである KafkaSourceReader メトリックグループに登録されます。特定のトピックパーティションに関連するメトリックは、KafkaSourceReader.topic.<topic_name>.partition.<partition_id> サブグループに登録されます。

      例えば、「my-topic」 トピックのパーティション 1 の現在のコンシューマーオフセットメトリック (currentOffset) は、.operator.KafkaSourceReader.topic.my-topic.partition.1.currentOffset で利用可能です。成功したコミット数 (commitsSucceeded) は、.operator.KafkaSourceReader.commitsSucceeded で利用可能です。

    • メトリックのリスト

      メトリック	説明	範囲
      currentOffset	パーティションの現在のコンシューマーオフセット。	TopicPartition
      committedOffset	パーティションの最後にコミットされたオフセット。	TopicPartition
      commitsSucceeded	成功したオフセットコミットの総数。	KafkaSourceReader
      commitsFailed	失敗したコミット数	KafkaSourceReader

    • Kafka コンシューマーメトリック

      基盤となる Kafka コンシューマーのメトリックは、KafkaSourceReader.KafkaConsumer メトリックグループに登録されます。たとえば、records-consumed-total メトリックは .operator.KafkaSourceReader.KafkaConsumer.records-consumed-total に登録されます。

      register.consumer.metrics プロパティを使用して、Kafka コンシューマーメトリックを登録するかどうかを指定できます。このオプションはデフォルトで有効になっています (true)。Kafka コンシューマーメトリックの詳細については、「Apache Kafka ドキュメント」をご参照ください。

• Kafka シンクの構築

  Flink Kafka シンクは、データストリームを 1 つ以上の Kafka トピックに書き込みます。

  DataStream<String> stream = ...
  
  Properties kafkaProperties = new Properties();
  kafkaProperties.setProperty("bootstrap.servers", "localhost:9092");
  
  KafkaSink<String> sink = KafkaSink.<String>builder()
          .setKafkaProducerConfig(kafkaProperties)
          .setRecordSerializer(
                  KafkaRecordSerializationSchema.builder()
                          .setTopic("my-topic")
                          .setValueSerializationSchema(new SimpleStringSchema())
                          .build())
          .setDeliveryGuarantee(DeliveryGuarantee.AT_LEAST_ONCE)
          .build();
  
  stream.sinkTo(sink);

  Kafka シンクを構築するには、次のプロパティを構成する必要があります。

  パラメーター	説明
  Kafka クライアントプロパティ	bootstrap.servers プロパティは必須です。これには、カンマ区切りの Kafka ブローカーのリストを指定します。
  レコードシリアライザー	入力データを Kafka の ProducerRecord に変換するには、KafkaRecordSerializationSchema を提供する必要があります。 Flink は、メッセージキーと値のシリアル化、トピックの選択、メッセージのパーティショニングなどの一般的なコンポーネントを提供するスキーマビルダーを提供します。 また、対応するインターフェイスを実装して、よりきめ細かな制御を行うことも可能です。 Kafka に書き込む ProducerRecord を生成するために、受信レコードごとに ProducerRecord<byte[], byte[]> serialize(T element, KafkaSinkContext context, Long timestamp) メソッドが呼び出されます。 ProducerRecord は、各レコードを Kafka に書き込む方法を詳細にコントロールし、次の操作を可能にします。 送信先トピックの設定。 メッセージのキーの設定。 送信先パーティションの指定。
  配信保証	bootstrap.servers パラメーターは必須であり、カンマ区切りの Kafka ブローカーのリストを指定します。
  配信保証	Flink のチェックポイントが有効な場合、Flink Kafka シンクは 1 回限りのセマンティクスを提供できます。チェックポイントを有効にすることに加えて、DeliveryGuarantee パラメーターを使用して異なる配信保証を指定できます。DeliveryGuarantee パラメーターは次のオプションを提供します： DeliveryGuarantee.NONE：(デフォルト) Flink は保証を提供しません。データが失われたり、重複したりする可能性があります。 DeliveryGuarantee.AT_LEAST_ONCE：データが失われないことを保証しますが、重複が発生する可能性があります。 DeliveryGuarantee.EXACTLY_ONCE：Kafka トランザクションを使用して 1 回限りのセマンティクスを提供します。 説明 EXACTLY_ONCE セマンティクスを使用する場合は、「1 回限りのセマンティクスに関する考慮事項」をご参照ください。

ダーティデータの許容範囲とコレクション

ご利用の Kafka データソースには、一般的にダーティデータと呼ばれる不正なレコードが含まれている場合があります。ジョブの度重なる失敗や再起動を防ぐため、これらの無効なレコードをスキップするように構成できます。以下に例を示します。

source:
  type: kafka
  name: Kafka Source
  properties.bootstrap.servers: host:9092
  topic: test-topic
  value.format: json
  scan.startup.mode: earliest-offset
  # Enable Dirty Data Tolerance
  ingestion.ignore-errors: true
  # Tolerate up to 1000 dirty data records
  ingestion.error-tolerance.max-count: 1000

この構成では、ジョブは 1,000 件以下のダーティレコードを検出する限り、実行を継続します。その数がしきい値を超えると、ジョブは失敗するため、データを調査できます。

ダーティデータによってジョブが決して失敗しないようにするには、以下の構成を使用します。

source:
  type: kafka
  name: Kafka Source
  properties.bootstrap.servers: host:9092
  topic: test-topic
  value.format: json
  scan.startup.mode: earliest-offset
  # Enable Dirty Data Tolerance
  ingestion.ignore-errors: true
  # Tolerate all dirty data records
  ingestion.error-tolerance.max-count: -1

ダーティデータの許容範囲を設定することでジョブは実行し続けますが、問題のあるレコードを検査したい場合もあるでしょう。また、ダーティデータを分析して Kafka プロデューサーを改善することもできます。「ダーティデータのコレクション」で説明されているように、TaskManager のログでジョブのダーティデータを確認できます。以下に例を示します。

source:
  type: kafka
  name: Kafka Source
  properties.bootstrap.servers: host:9092
  topic: test-topic
  value.format: json
  scan.startup.mode: earliest-offset
  # Enable Dirty Data Tolerance
  ingestion.ignore-errors: true
  # Tolerate all dirty data records
  ingestion.error-tolerance.max-count: -1

pipeline:
  dirty-data.collector:
    # Write dirty data to the TaskManager log file
    type: logger

テーブル名とトピックのマッピング

Kafka が Flink CDC の sink として機能する場合、メッセージフォーマット (Debezium JSON や Canal JSON など) には元のテーブル名が埋め込まれます。下流のコンシューマーは通常、トピック名ではなく、この埋め込まれた名前をテーブル識別子として使用するため、テーブル名とトピック間のマッピングを正しく設定することが重要です。

MySQL データベースから 2 つのテーブル、mydb.mytable1 と mydb.mytable2 を同期する必要があるとします。次のマッピング戦略を利用できます：

1. マッピング戦略なし

マッピング戦略がない場合、各テーブルのデータは「<データベース名>.<テーブル名>」というフォーマットで命名されたトピックに書き込まれます。したがって、mydb.mytable1 のデータは mydb.mytable1 という名前のトピックに、mydb.mytable2 のデータは mydb.mytable2 という名前のトピックに書き込まれます。以下に構成例を示します：

source:
  type: mysql
  name: MySQL Source
  hostname: ${secret_values.mysql.hostname}
  port: ${mysql.port}
  username: ${secret_values.mysql.username}
  password: ${secret_values.mysql.password}
  tables: mydb.mytable1,mydb.mytable2
  server-id: 8601-8604

sink:
  type: kafka
  name: Kafka Sink
  properties.bootstrap.servers: ${kafka.bootstraps.server}

2. ルートルールによるマッピング (非推奨)

デフォルトの「<データベース名>.<テーブル名>」フォーマットを使用する代わりに、特定のトピックにデータを書き込みたい場合があります。その場合、ルートルールを設定できます。以下に構成例を示します：

source:
  type: mysql
  name: MySQL Source
  hostname: ${secret_values.mysql.hostname}
  port: ${mysql.port}
  username: ${secret_values.mysql.username}
  password: ${secret_values.mysql.password}
  tables: mydb.mytable1,mydb.mytable2
  server-id: 8601-8604

sink:
  type: kafka
  name: Kafka Sink
  properties.bootstrap.servers: ${kafka.bootstraps.server}
  
 route:
  - source-table: mydb.mytable1,mydb.mytable2
    sink-table: mytable

この場合、mydb.mytable1 と mydb.mytable2 からのすべてのデータは、mytable という名前の単一のトピックに書き込まれます。

ただし、送信先トピックを変更するルートルールは、Kafka メッセージ (Debezium JSON または Canal JSON フォーマット) 内のテーブル名も変更してしまいます。すべての Kafka メッセージ内のテーブル名が mytable となり、このトピックからメッセージを消費するシステムで予期しない動作を引き起こす可能性があります。

3. sink.tableId-to-topic.mapping によるマッピング (推奨)

元のソーステーブル名を保持したままテーブル名をトピックにマッピングするには、`sink.tableId-to-topic.mapping` パラメーターを使用します。以下に構成例を示します：

source:
  type: mysql
  name: MySQL Source
  hostname: ${secret_values.mysql.hostname}
  port: ${mysql.port}
  username: ${secret_values.mysql.username}
  password: ${secret_values.mysql.password}
  tables: mydb.mytable1,mydb.mytable2
  server-id: 8601-8604
  sink.tableId-to-topic.mapping: mydb.mytable1,mydb.mytable2:mytable

sink:
  type: kafka
  name: Kafka Sink
  properties.bootstrap.servers: ${kafka.bootstraps.server}

または、以下の構成を使用することもできます：

source:
  type: mysql
  name: MySQL Source
  hostname: ${secret_values.mysql.hostname}
  port: ${mysql.port}
  username: ${secret_values.mysql.username}
  password: ${secret_values.mysql.password}
  tables: mydb.mytable1,mydb.mytable2
  server-id: 8601-8604
  sink.tableId-to-topic.mapping: mydb.mytable1:mytable;mydb.mytable2:mytable

sink:
  type: kafka
  name: Kafka Sink
  properties.bootstrap.servers: ${kafka.bootstraps.server}

この場合、mydb.mytable1 と mydb.mytable2 からのすべてのデータは mytable トピックに書き込まれ、Kafka メッセージ (Debezium JSON または Canal JSON フォーマット) 内のテーブル名は mydb.mytable1 または mydb.mytable2 として保持されます。これにより、下流のシステムは各レコードの元のソーステーブルを識別し続けることができます。

設定例

以下の例では、一般的なユースケースの設定を示します。

source:
  type: kafka
  topic: customers
  properties.bootstrap.servers: localhost:9092
  properties.group.id: ${kafka.group.id}
  value.format: json
  # オプション: 各レコードのスキーマを動的に推論し、スキーマの変更を検出します。
  schema.inference.strategy: continuous

sink:
  type: paimon
  name: Paimon Sink
  catalog.properties.metastore: rest
  catalog.properties.uri: dlf_uri
  catalog.properties.warehouse: your_warehouse
  catalog.properties.token.provider: dlf
  # オプション: コミットユーザーを指定します。競合を防ぐために、ジョブごとに異なるコミットユーザーを使用することを推奨します。
  commit.user: your_job_name
  # オプション: 削除ベクターを有効にして、読み取りパフォーマンスを向上させます。
  table.properties.deletion-vectors.enabled: true

source:
  type: kafka
  topic-pattern: user_event_.*
  properties.bootstrap.servers: localhost:9092
  properties.group.id: ${kafka.group.id}
  value.format: json
  # オプション: 各レコードのスキーマを動的に推論し、スキーマの変更を検出します。
  schema.inference.strategy: continuous

sink:
  type: starrocks
  jdbc-url: jdbc:mysql://<yourFeHostname>:9030
  load-url: <yourFeHostname>:8030
  username: <yourUsername>
  password: ${secret_values.starrocks_password}
 
  # オプション: データ量が少ないジョブの場合、フラッシュ間隔を短くして、データが長時間書き込まれないままになるのを防ぎます。デフォルトは 300000 ミリ秒 (5 分) です。
  sink.buffer-flush.interval-ms: 5000
  # オプション: アップストリームの文字セットが utf8mb4 の場合、テキストの切り捨てを防ぐためにこのパラメーターを 4 に設定します。デフォルトは 3 です。
  unicode-char.max-bytes: 4
  # オプション: 自動作成されるテーブルのバケット数を指定します。StarRocks バージョン 2.5.7 より前では、このパラメーターを明示的に設定する必要があります。それ以降のバージョンでは値を推論できます。
  table.create.num-buckets: 8
  # オプション: クラスター構成に基づいて、自動作成されるテーブルのレプリカ数を指定します。
  table.create.properties.replication_num: 3
  # オプション: StarRocks 3.2 以降では、この機能を有効にしてスキーマ変更を高速化します。
  table.create.properties.fast_schema_evolution: true
  # 注: 変換によってプライマリキーが変更された場合は、sink.ignore.update-before も false に設定する必要があります。
  # そうしないと、古いプライマリキーを使用する行がダウンストリームシステムに残ってしまいます。

source:
  type: kafka
  topic: customers
  properties.bootstrap.servers: localhost:9092
  properties.group.id: ${kafka.group.id}
  value.format: json
  # オプション: 各レコードのスキーマを動的に推論し、スキーマの変更を検出します。
  schema.inference.strategy: continuous
  metadata.list: topic,partition

sink:
  type: paimon
  name: Paimon Sink
  catalog.properties.metastore: rest
  catalog.properties.uri: dlf_uri
  catalog.properties.warehouse: your_warehouse
  catalog.properties.token.provider: dlf
  # オプション: コミットユーザーを指定します。競合を防ぐために、ジョブごとに異なるコミットユーザーを使用することを推奨します。
  commit.user: your_job_name
  # オプション: 削除ベクターを有効にして、読み取りパフォーマンスを向上させます。
  table.properties.deletion-vectors.enabled: true

source:
  type: kafka
  topic: customers
  properties.bootstrap.servers: localhost:9092
  properties.group.id: ${kafka.group.id}
  value.format: json
  # オプション: 各レコードのスキーマを動的に推論し、スキーマの変更を検出します。
  schema.inference.strategy: continuous
  # 解析エラーを無視します。この機能を有効にすると、デフォルトですべての解析エラーが無視されます。
  ingestion.ignore-errors: true

sink:
  type: paimon
  name: Paimon Sink
  catalog.properties.metastore: rest
  catalog.properties.uri: dlf_uri
  catalog.properties.warehouse: your_warehouse
  catalog.properties.token.provider: dlf
  # オプション: コミットユーザーを指定します。競合を防ぐために、ジョブごとに異なるコミットユーザーを使用することを推奨します。
  commit.user: your_job_name
  # オプション: 削除ベクターを有効にして、読み取りパフォーマンスを向上させます。
  table.properties.deletion-vectors.enabled: true
  
# ダーティデータコレクターを有効にして、解析できないレコードを出力します。
pipeline:
  dirty-data.collector:
    name: Logger Dirty Data Collector
    type: logger

source:
  type: kafka
  topic: customers
  properties.bootstrap.servers: localhost:9092
  properties.group.id: ${kafka.group.id}
  value.format: json
  # オプション: 各レコードのスキーマを動的に推論し、スキーマの変更を検出します。
  schema.inference.strategy: continuous
  # 解析エラーを無視します。
  ingestion.ignore-errors: true
  # 30 回の解析エラーが発生した後にジョブを失敗させます。
  ingestion.error-tolerance.max-count: 30
  
sink:
  type: paimon
  name: Paimon Sink
  catalog.properties.metastore: rest
  catalog.properties.uri: dlf_uri
  catalog.properties.warehouse: your_warehouse
  catalog.properties.token.provider: dlf
  # オプション: コミットユーザーを指定します。競合を防ぐために、ジョブごとに異なるコミットユーザーを使用することを推奨します。
  commit.user: your_job_name
  # オプション: 削除ベクターを有効にして、読み取りパフォーマンスを向上させます。
  table.properties.deletion-vectors.enabled: true
  
# ダーティデータコレクターを有効にして、解析できないレコードを出力します。
pipeline:
  dirty-data.collector:
    name: Logger Dirty Data Collector
    type: logger

source:
  type: kafka
  topic: customers
  properties.bootstrap.servers: localhost:9092
  properties.group.id: ${kafka.group.id}
  value.format: json
  # オプション: 各レコードのスキーマを動的に推論し、スキーマの変更を検出します。
  schema.inference.strategy: continuous
  # db フィールドと tbl フィールドをテーブル ID として使用します。
  value.json.decode.parser-table-id.fields: db,tbl

sink:
  type: paimon
  name: Paimon Sink
  catalog.properties.metastore: rest
  catalog.properties.uri: dlf_uri
  catalog.properties.warehouse: your_warehouse
  catalog.properties.token.provider: dlf
  # オプション: コミットユーザーを指定します。競合を防ぐために、ジョブごとに異なるコミットユーザーを使用することを推奨します。
  commit.user: your_job_name
  # オプション: 削除ベクターを有効にして、読み取りパフォーマンスを向上させます。
  table.properties.deletion-vectors.enabled: true
  
# ダーティデータコレクターを有効にして、解析できないレコードを出力します。
pipeline:
  dirty-data.collector:
    name: Logger Dirty Data Collector
    type: logger

source:
  type: kafka
  topic: customers
  properties.bootstrap.servers: localhost:9092
  properties.group.id: ${kafka.group.id}
  value.format: json
  # db フィールドと tbl フィールドをテーブル ID として使用します。
  value.json.decode.parser-table-id.fields: db,tbl
  # 選択したフィールドに固定タイプを指定します。
  value.json.infer-schema.fixed-types: 'db STRING, tbl STRING, id BIGINT, amount DECIMAL(18, 2)'
  # 宣言されていないフィールドは引き続き動的に推論します。
  schema.inference.strategy: continuous

sink:
  type: paimon
  name: Paimon Sink
  catalog.properties.metastore: rest
  catalog.properties.uri: dlf_uri
  catalog.properties.warehouse: your_warehouse
  catalog.properties.token.provider: dlf
  # オプション: コミットユーザーを指定します。競合を防ぐために、ジョブごとに異なるコミットユーザーを使用することを推奨します。
  commit.user: your_job_name
  # オプション: 削除ベクターを有効にして、読み取りパフォーマンスを向上させます。
  table.properties.deletion-vectors.enabled: true
  
# ダーティデータコレクターを有効にして、解析できないレコードを出力します。
pipeline:
  dirty-data.collector:
    name: Logger Dirty Data Collector
    type: logger

source:
  type: kafka
  topic: customers
  properties.bootstrap.servers: localhost:9092
  properties.group.id: ${kafka.group.id}
  value.format: canal-json
  # オプション: 各レコードのスキーマを動的に推論し、スキーマの変更を検出します。
  schema.inference.strategy: continuous

sink:
  type: paimon
  name: Paimon Sink
  catalog.properties.metastore: rest
  catalog.properties.uri: dlf_uri
  catalog.properties.warehouse: your_warehouse
  catalog.properties.token.provider: dlf
  # オプション: コミットユーザーを指定します。競合を防ぐために、ジョブごとに異なるコミットユーザーを使用することを推奨します。
  commit.user: your_job_name
  # オプション: 削除ベクターを有効にして、読み取りパフォーマンスを向上させます。
  table.properties.deletion-vectors.enabled: true
  
# ダーティデータコレクターを有効にして、解析できないレコードを出力します。
pipeline:
  dirty-data.collector:
    name: Logger Dirty Data Collector
    type: logger

source:
  type: kafka
  topic: customers
  properties.bootstrap.servers: localhost:9092
  properties.group.id: ${kafka.group.id}
  value.format: canal-json
  # オプション: 各レコードのスキーマを動的に推論し、スキーマの変更を検出します。
  schema.inference.strategy: continuous
  # MySQL 型情報からスキーマを推論します。このパラメーターを SQL_TYPE に設定して、代わりに SQL 型情報を使用することもできます。
  value.canal-json.infer-schema.strategy: MYSQL_TYPE

sink:
  type: paimon
  name: Paimon Sink
  catalog.properties.metastore: rest
  catalog.properties.uri: dlf_uri
  catalog.properties.warehouse: your_warehouse
  catalog.properties.token.provider: dlf
  # オプション: コミットユーザーを指定します。競合を防ぐために、ジョブごとに異なるコミットユーザーを使用することを推奨します。
  commit.user: your_job_name
  # オプション: 削除ベクターを有効にして、読み取りパフォーマンスを向上させます。
  table.properties.deletion-vectors.enabled: true
  
# ダーティデータコレクターを有効にして、解析できないレコードを出力します。
pipeline:
  dirty-data.collector:
    name: Logger Dirty Data Collector
    type: logger

source:
  type: kafka
  topic: customers
  properties.bootstrap.servers: localhost:9092
  properties.group.id: ${kafka.group.id}
  value.format: debezium-json
  # オプション: 各レコードのスキーマを動的に推論し、スキーマの変更を検出します。
  schema.inference.strategy: continuous

sink:
  type: paimon
  name: Paimon Sink
  catalog.properties.metastore: rest
  catalog.properties.uri: dlf_uri
  catalog.properties.warehouse: your_warehouse
  catalog.properties.token.provider: dlf
  # オプション: コミットユーザーを指定します。競合を防ぐために、ジョブごとに異なるコミットユーザーを使用することを推奨します。
  commit.user: your_job_name
  # オプション: 削除ベクターを有効にして、読み取りパフォーマンスを向上させます。
  table.properties.deletion-vectors.enabled: true

source:
  type: mysql
  hostname: #{hostname}
  port: 3306
  username: #{username}
  password: #{password}
  tables: order_dw.\.*
  server-id: 28601-28604
  # オプション: 増分フェーズで作成されたテーブルからデータを同期します。
  scan.binlog.newly-added-table.enabled: true
  # オプション: テーブルと列のコメントを同期します。
  include-comments.enabled: true
  # オプション: TaskManager のメモリ不足エラーの可能性を防ぐために、まず非有界チャンクを処理します。
  scan.incremental.snapshot.unbounded-chunk-first.enabled: true
  # オプション: 解析フィルターを有効にして読み取りを高速化します。
  scan.only.deserialize.captured.tables.changelog.enabled: true
  # mysqlType、sqlType、sql、isDdl、およびその他のメタデータを Canal JSON データに追加します。
  include-binlog-meta.enable: true
  
sink:
  type: kafka
  properties.bootstrap.servers: localhost:9092
  topic: order_dw_tables
  # Kafka の値に Canal JSON チェンジログフォーマットを使用します。
  value.format: canal-json
  # 日付と時刻データのシリアル化に使用するフォーマットを指定します。
  value.canal-json.timestamp-format.standard: SQL
  # binlog の順序を維持するために、すべてのデータをパーティション 0 に書き込みます。
  partition.strategy: all-to-zero

Exactly-once セマンティクス

• コンシューマーの分離レベルの設定

  Kafka データを消費するすべてのアプリケーションでは、isolation.level プロパティを設定する必要があります。

  • read_committed：コミット済みのデータのみを読み取ります。

  • read_uncommitted (デフォルト)：コミットされていないデータも読み取ることができます。

  EXACTLY_ONCE は read_committed に依存します。そうでない場合、コンシューマーがコミットされていないデータを読み取り、一貫性が損なわれる可能性があります。

• トランザクションタイムアウトとデータ損失

  チェックポイントから回復する際、Realtime Compute for Apache Flink は、そのチェックポイントが開始される前にコミットされたトランザクションのみを考慮します。ジョブの障害発生から再起動までの時間が Kafka のトランザクションタイムアウトを超えると、Kafka は開いているトランザクションを自動的に中止し、データ損失が発生する可能性があります。

  • Kafka ブローカーの transaction.max.timeout.ms のデフォルト値は 15 分です。

  • デフォルトでは、Flink Kafka Sink は transaction.timeout.ms パラメーターを 1 時間に設定します。

  • ブローカーの transaction.max.timeout.ms を、Flink での設定値以上に増やす必要があります。

• プロデューサープールと同時チェックポイント

  EXACTLY_ONCE モードでは、固定サイズの Kafka プロデューサープールが使用されます。各チェックポイントは、このプールから 1 つのプロデューサーを使用します。同時チェックポイントの数がプールサイズを超えると、ジョブは失敗します。

  同時チェックポイントの最大数に基づいて、プロデューサープールのサイズを設定してください。

• 並列度のスケールダウンに関する制約

  最初のチェックポイントが完了する前にジョブが失敗した場合、再起動時に元のプロデューサープールの情報が失われます。そのため、最初のチェックポイントが完了するまでは、ジョブの並列度をスケールダウンしないでください。スケールダウンが必要な場合でも、新しい並列度は FlinkKafkaProducer.SAFE_SCALE_DOWN_FACTOR 未満にすることはできません。

• トランザクションによる読み取りのブロック

  read_committed モードでは、コミットも中止もされていないトランザクションがあると、トピック全体の読み取り操作がブロックされます。

  例：

  • トランザクション 1 がデータを書き込みます。

  • トランザクション 2 がさらにデータを書き込み、コミットされます。

  • トランザクション 1 が開いている限り、コミット済みのトランザクション 2 のデータはコンシューマーには表示されません。

  これには、以下の影響があります。

  • 通常の操作中、データが表示されるまでのレイテンシーは、おおよそチェックポイントの間隔と等しくなります。

  • ジョブが失敗した場合、そのジョブが書き込んでいたトピックは、ジョブが再起動するかトランザクションがタイムアウトするまで、コンシューマーに対してブロックされます。極端なケースでは、トランザクションのタイムアウト処理自体が読み取り操作に影響を与えることもあります。

よくある質問

• Realtime Compute for Apache Flink を使用して Kafka から JSON データを取得するにはどうすればよいですか？

• ApsaraMQ for Kafka のソーステーブルに対して、イベント時間ベースのウィンドウが出力されないのはなぜですか？

• 保護された Kafka クラスターに接続するにはどうすればよいですか？

• フィールド名の競合を解決するにはどうすればよいですか？

• ワークスペースと名前空間