Realtime Compute for Apache Flink:Message Queue for Apache Kafka

背景情報

前提条件

注意事項

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

• 各チェックポイントごとに新しいトランザクション ID が生成されます。チェックポイント間隔が短すぎると、過剰な数のトランザクション ID が作成されます。これにより、Apache Kafka クラスター内のコーディネーターのメモリが不足し、クラスターの安定性が損なわれる可能性があります。

• 各トランザクションごとに Producer インスタンスが作成されます。同時に多数のトランザクションがコミットされると、TaskManager のメモリが不足し、Apache Flink ジョブの安定性が損なわれる可能性があります。

• 複数の Apache Flink ジョブが同じ sink.transactional-id-prefix を使用している場合、生成されたトランザクション ID が競合する可能性があります。あるジョブでの書き込み操作が失敗すると、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

Kafka コネクタは、SQL ジョブにおけるソーステーブルまたは結果テーブルとして使用できます。

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',
  ...
);

WITH パラメーター

• 一般

  パラメーター	説明	タイプ	必須	デフォルト	備考
  connector	使用するコネクタ。	String	はい	該当なし	値は kafka でなければなりません。
  properties.bootstrap.servers	Kafka ブローカーのアドレス一覧。	String	はい	該当なし	書式： host:port,host:port,...。アドレスはカンマ (,) で区切ります。
  properties.*	Kafka クライアントの追加プロパティ。	String	いいえ	該当なし	プロパティキーは、公式 Apache Kafka ドキュメントのProducer Configs およびConsumer Configs で定義された有効なオプションである必要があります。 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 の両方を設定することはできません。両方が構成されている場合、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	いいえ	該当なし	説明 このオプションと topic のどちらか一方のみを指定できます。両方を指定することはできません。
  properties.group.id	Kafka ソースのコンシューマーグループ ID。	String	いいえ	KafkaSource-{ソーステーブル名}	初めてコンシューマーグループ ID を使用する場合、初期スタートオフセットを定義するために、properties.auto.offset.reset を earliest または latest のいずれかに設定する必要があります。
  scan.startup.mode	Kafka コンシューマーのスタートオフセット。	String	いいえ	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 の場合の各パーティションのスタートオフセット。	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	並列 sink インスタンスから Kafka パーティションへのレコードのマッピング方法を定義します。	String	いいえ	default	有効な値： default: Kafka のデフォルトパーティショナーを使用します。 fixed: 各並列 sink インスタンスが固定の Kafka パーティションに書き込みます。 round-robin: レコードはラウンドロビン方式でパーティションに配布されます。 カスタムパーティショナー：カスタムパーティショナーを使用するには、FlinkKafkaPartitioner サブクラスの完全修飾クラス名（例：org.mycompany.MyPartitioner）を指定します。
  sink.delivery-guarantee	sink の配信保証レベル。	String	いいえ	at-least-once	有効な値： none: 保証は提供されません。レコードが失われるか、重複する可能性があります。 at-least-once: レコードが失われることはありませんが、重複する可能性があります。 exactly-once: Kafka トランザクションを使用して「1 回限りのセマンティクス」を提供し、レコードが失われず、重複しないことを保証します。 説明 exactly-once セマンティクスを使用する場合、sink.transactional-id-prefix も指定する必要があります。
  sink.transactional-id-prefix	sink.delivery-guarantee が exactly-once の場合に必要なトランザクション ID のプレフィックス。	String	はい。sink.delivery-guarantee が exactly-once の場合です。	該当なし	sink.delivery-guarantee を exactly-once に設定した場合にのみ必須です。
  sink.parallelism	sink オペレーターの並列度。	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 証明書の truststore のパス。 */
  /* Artifacts を使用してアップロードされたファイルは /flink/usrlib/ ディレクトリに格納されます。 */
  'properties.ssl.truststore.location' = '/flink/usrlib/kafka.client.truststore.jks',
  'properties.ssl.truststore.password' = 'test1234',
  /* クライアント認証が必要な場合、keystore（秘密鍵）のパスも構成する必要があります。 */
  '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";'
)

Real-time Compute コンソールの Artifacts 機能を使用して、上記の例で言及されている CA 証明書および秘密鍵をアップロードできます。アップロードされたファイルは /flink/usrlib ディレクトリに格納されます。my-truststore.jks という名前の CA 証明書ファイルを使用する場合、WITH 句の 'properties.ssl.truststore.location' プロパティを、以下のいずれかの方法で設定できます。

• 'properties.ssl.truststore.location' = '/flink/usrlib/my-truststore.jks' を設定します。この方法では、Object Storage Service (OSS) から実行時にファイルを動的にダウンロードする必要がなくなりますが、デバッグモードはサポートされません。

• Realtime Compute エンジンのバージョンが 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 クラスター内のコンシューマーオフセットが遅延します。チェックポイント中、Kafka ソーステーブルは現在の読み取り進行状況をその状態に保存します。システムはこの状態を使用して障害復旧を行い、Kafka クラスターにコミットされたオフセットには依存しません。オフセットは、Kafka 内での読み取り進行状況の監視のみを目的としてコミットされます。オフセットのコミット失敗は、データの正確性には影響しません。

カスタム sink パーティショナー

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

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

Kafka は、データの更新または削除をサポートしない追加専用のメッセージキューです。ストリーミング SQL では、標準の Kafka 結果テーブルは、上流の Change Data Capture (CDC) データや集計および結合などのオペレーターの再tractionロジックを処理できません。変更または再tractionを含むデータを書き込む必要がある場合は、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 ソース は、Kafka ソース インスタンスを作成するためのビルダークラスを提供します。次のサンプルコードは、input-topic トピック の最も古い オフセット からデータを消費する Kafka ソース を構築します。コンシューマーグループ は 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) メソッドを呼び出して設定します。
  Topics または Partitions	サブスクライブ するトピックまたはパーティション。この 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 "topic-a" のパーティション 0 new TopicPartition("topic-b", 5))); // topic "topic-b" のパーティション 5 KafkaSource.builder().setPartitions(partitionSet)
  Deserializer	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 セントラルリポジトリで利用可能です。

  <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 ソース には通常、停止 オフセット がありません。ただし、テスト目的で、ストリーミングモード でも setUnbounded(OffsetsInitializer) を使用して停止 オフセット を指定できます。ストリーミングモード と バッチモード で停止 オフセット を指定するメソッド名は異なります：ストリーミングモード では setUnbounded、バッチモード では setBounded です。

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

    Flink ジョブ を再起動することなく、トピックパターンによるサブスクライブ時にトピックのスケーリングまたは新規トピックの作成を処理するには、トピック のスケーリングに対応するために 動的パーティション検出 を有効化できます。この機能はデフォルトで無効化されており、明示的に有効化する必要があります。

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

    重要

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

  • イベント時刻および Watermark

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

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

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

    説明

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

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

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

    • 適切な 並列度 の設定：ソースの 並列度 が 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 によって提供される戦略によって上書きされます。

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

    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 の KafkaRecordSerializationSchema を使用して Kafka の ProducerRecord に変換する必要があります。Flink では、メッセージキーおよび値のシリアル化、トピック選択、メッセージのパーティション分割など、一般的なコンポーネントを提供するスキーマビルダーが用意されています。さらに細かい制御が必要な場合は、対応するインターフェイスを実装することもできます。各受信レコードに対して ProducerRecord<byte[], byte[]> serialize(T element, KafkaSinkContext context, Long timestamp) メソッドが呼び出され、Kafka に書き込む ProducerRecord を生成します。 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 セマンティクスを使用する場合は、「EXACTLY_ONCE セマンティクスに関する考慮事項」をご参照ください。

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

ご利用の Kafka データソースには、不正な形式のレコード（いわゆる「ダーティデータ」）が含まれている可能性があります。ジョブが頻繁に失敗・再起動するのを防ぐため、無効なレコードを無視するよう構成できます。たとえば、以下のとおりです。

source:
  type: kafka
  name: Kafka Source
  properties.bootstrap.servers: host:9092
  topic: test-topic
  value.format: json
  scan.startup.mode: earliest-offset
  # ダーティデータの許容を有効化
  ingestion.ignore-errors: true
  # 最大 1,000 件のダーティデータレコードを許容
  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
  # ダーティデータの許容を有効化
  ingestion.ignore-errors: true
  # 全てのダーティデータレコードを許容
  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
  # ダーティデータの許容を有効化
  ingestion.ignore-errors: true
  # 全てのダーティデータレコードを許容
  ingestion.error-tolerance.max-count: -1

pipeline:
  dirty-data.collector:
    # ダーティデータを TaskManager のログファイルに書き込む
    type: logger

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

データ取り込みジョブの Sink として Kafka を使用する場合、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 として保持されます。これにより、ダウンストリームシステムは元のソーステーブル名を正しく取得できます。

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 データを取得する方法は？

• Event Time を基準としたウィンドウで、Message Queue for Apache Kafka のソーステーブルから出力が得られない理由は？

• セキュアな Kafka クラスターに接続する方法は？

• フィールド名の競合を解決する方法は？

• ワークスペースとネームスペース