Realtime Compute for Apache Flink:ApsaraMQ for Kafka

背景情報

前提条件

注意事項

現在、Flink と Kafka の設計上のバグのため、トランザクション書き込みの使用は推奨されません。sink.delivery-guarantee = exactly-once を設定すると、Kafka コネクタはトランザクション書き込みを有効にしますが、3 つの既知の問題があります：

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

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

• 複数の Flink ジョブが同じ sink.transactional-id-prefix を使用すると、生成されるトランザクション ID が競合する可能性があります。1 つのジョブの書き込みに失敗すると、Kafka パーティションの Log Start Offset (LSO) の進行がブロックされます。これにより、そのパーティションからデータを読み取るすべてのコンシューマーに影響が及びます。

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

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

Flink ジョブの起動中に Timed out waiting for a node assignment というエラーが報告された場合、通常は 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,host:port です。アドレスはコンマ (,) で区切ります。
  properties.*	Kafka クライアントの直接設定。	String	いいえ	なし	サフィックスは、Kafka の公式ドキュメントで定義されているプロデューサーおよびコンシューマーの設定である必要があります。 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	Kafka メッセージのすべてのキーフィールドにカスタムプレフィックスを指定して、メッセージの値部分のフィールドとの名前の競合を回避します。	String	いいえ	なし	この設定項目は、ソーステーブルと結果テーブルの列名を区別するためにのみ使用されます。Kafka メッセージのキー部分を解析および生成する際にプレフィックスは削除されます。 説明 この設定を使用する場合、value.fields-include を EXCEPT_KEY に設定する必要があります。
  value.format	Kafka メッセージの値部分を読み書きするために使用されるフォーマット。	String	いいえ	なし	この設定は format と同等です。format または value.format のいずれか一方のみを設定できます。両方が設定されている場合、value.format が format を上書きします。
  value.fields-include	Kafka メッセージの値部分を解析または生成する際に、メッセージのキー部分に対応するフィールドを含めるかどうかを指定します。	String	いいえ	ALL	有効な値: ALL (デフォルト): すべての列が Kafka メッセージの値部分として処理されます。 EXCEPT_KEY: key.fields で定義されたものを除く残りのフィールドが、Kafka メッセージの値部分として処理されます。

• ソーステーブル

  パラメーター	説明	データ型	必須	デフォルト値	備考
  topic	読み取るトピックの名前。	String	いいえ	なし	複数のトピック名はセミコロン (;) で区切ります。例: topic-1;topic-2。 説明 topic と topic-pattern オプションのいずれか一方のみを指定できます。
  topic-pattern	読み取り元のトピック名に一致する正規表現。ジョブの実行中に、この正規表現に一致するすべてのトピックが読み取られます。	String	いいえ	なし	説明 topic と topic-pattern オプションのいずれか一方のみを指定できます。
  properties.group.id	コンシューマーグループ ID。	String	いいえ	KafkaSource-{source_table_name}	指定されたグループ ID を初めて使用する場合は、properties.auto.offset.reset を earliest または latest に設定して、初期開始オフセットを指定する必要があります。
  scan.startup.mode	Kafka からデータを読み取るための開始オフセット。	String	いいえ	group-offsets	有効な値: earliest-offset：Kafka の最も早いパーティションから読み取りを開始します。 latest-offset：Kafka の最新のオフセットから読み取りを開始します。 group-offsets (デフォルト)：指定された properties.group.id のコミット済みオフセットから読み取りを開始します。 timestamp：scan.startup.timestamp-millis で指定されたタイムスタンプから読み取りを開始します。 specific-offsets：scan.startup.specific-offsets で指定されたオフセットから読み取りを開始します。 説明 このパラメーターは、状態なしでタスクが開始された場合に有効になります。タスクがチェックポイントから再開するか、状態から回復すると、状態に保存された進行状況を優先的に使用して読み取りを再開します。
  scan.startup.specific-offsets	specific-offsets スタートアップモードで、各パーティションの開始オフセットを指定します。	String	いいえ	なし	例: partition:0,offset:42;partition:1,offset:300
  scan.startup.timestamp-millis	timestamp スタートアップモードで、開始オフセットのタイムスタンプを指定します。	Long	いいえ	なし	単位はミリ秒です。
  scan.topic-partition-discovery.interval	Kafka のトピックとパーティションを動的に検出する間隔。	Duration	いいえ	5 分	デフォルトのパーティションチェック間隔は 5 分です。この機能を無効にするには、パーティションチェック間隔を明示的に 0 以下の値に設定する必要があります。動的パーティション検出が有効な場合、Kafka ソースは新しいパーティションを自動的に検出し、そこからデータを読み取ることができます。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: タスクはコンシューマーグループの競合をチェックせずに直接開始します。 説明 このパラメーターは、VVR 6.0.4 以降でのみサポートされます。

• シンクテーブル

  パラメーター	説明	データ型	必須	デフォルト値	備考
  topic	書き込むトピックの名前。	String	はい	なし	なし
  sink.partitioner	Flink の並列処理から Kafka パーティションへのマッピングモード。	String	いいえ	default	有効な値: default: デフォルトの Kafka partitioner を使用します。 fixed: 各 Flink の並列処理は、固定の Kafka パーティションに対応します。 round-robin：Flink の並列度からのデータがラウンドロビン方式で Kafka パーティションに割り当てられます。 カスタム partitioner：fixed と round-robin がニーズに合わない場合は、FlinkKafkaPartitioner の子クラスを作成してカスタム partitioner を定義できます。例：org.mycompany.MyPartitioner
  sink.delivery-guarantee	Kafka 結果テーブルの配信セマンティクス。	String	いいえ	at-least-once	有効な値: none: 保証なし。データが失われたり、重複したりする可能性があります。 at-least-once (デフォルト): データの損失がないことを保証しますが、データが重複する可能性があります。 exactly-once: Kafka トランザクションを使用して、データが失われたり重複したりしないことを保証します。 説明 exactly-once セマンティクスを使用する場合、sink.transactional-id-prefix パラメーターが必要です。
  sink.transactional-id-prefix	exactly-once セマンティクスで使用される Kafka トランザクション ID のプレフィックス。	String	いいえ	なし	この設定は、sink.delivery-guarantee が exactly-once に設定されている場合にのみ有効です。
  sink.parallelism	Kafka 結果テーブルオペレーターの並列処理の次数。	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";'
)

例で言及されている CA 証明書と秘密鍵は、Realtime Compute コンソールのファイル管理機能を使用してプラットフォームにアップロードできます。アップロード後、ファイルは /flink/usrlib ディレクトリに保存されます。使用する CA 証明書ファイルの名前が my-truststore.jks の場合、WITH パラメーターで 'properties.ssl.truststore.location' = '/flink/usrlib/my-truststore.jks' を指定してこの証明書を使用します。

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

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 側での読み取り進捗の監視のためだけです。オフセットのコミットに失敗しても、データの正確性には影響しません。

結果テーブルのカスタム partitioner

組み込みの Kafka プロデューサー partitioner がニーズに合わない場合は、カスタム partitioner を実装して対応するパーティションにデータを書き込むことができます。カスタム partitioner は FlinkKafkaPartitioner を継承する必要があります。開発後、JAR パッケージをコンパイルし、Realtime Compute コンソールのファイル管理機能を使用してアップロードします。ファイルをアップロードして参照した後、WITH 句の sink.partitioner パラメーターを partitioner の完全なクラスパス (例：org.mycompany.MyPartitioner) に設定します。

Kafka、Upsert Kafka、Kafka JSON カタログの選択

Kafka は追記専用のメッセージキューシステムであり、データの更新や削除をサポートしていません。そのため、アップストリームの Change Data Capture (CDC) データや、ストリーミング SQL の集約や結合などのオペレーターからのリトラクションロジックを処理できません。変更やリトラクションを含むデータを Kafka に書き込むには、変更データを処理するために特別に設計されたUpsert Kafka 結果テーブルを使用します。

アップストリームデータベースの 1 つ以上のテーブルから Kafka に変更データをバッチで便利に同期するには、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>',  // Kafka ブローカーのリスト
  'properties.group.id' = '<yourKafkaConsumerGroupId>', // Kafka コンシューマーグループ ID
  'format' = 'csv'
);

CREATE TEMPORARY TABLE kafka_sink (
  id INT,
  name STRING,
  age INT
) WITH (
  'connector' = 'kafka',
  'topic' = 'sink',
  'properties.bootstrap.servers' = '<yourKafkaBrokers>', // Kafka ブローカーのリスト
  'properties.group.id' = '<yourKafkaConsumerGroupId>', // Kafka コンシューマーグループ ID
  '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>', // Kafka ブローカーのリスト
  '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 ソースは、KafkaSource インスタンスを作成するためのビルダー クラスを提供します。次のコード例は、`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");

  KafkaSource を構築する際には、次のパラメーターを指定する必要があります。

  パラメーター	説明
  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), // Partition 0 of topic "topic-a" new TopicPartition("topic-b", 5))); // Partition 5 of topic "topic-b" KafkaSource.builder().setPartitions(partitionSet)
  デシリアライザ	Kafka メッセージを解析するための逆シリアライザー。 setDeserializer(KafkaRecordDeserializationSchema) を使用してデシリアライザーを指定します。ここで、KafkaRecordDeserializationSchema は Kafka ConsumerRecord を解析する方法を定義します。Kafka メッセージのメッセージ本文 (値) のデータのみを解析する必要がある場合は、次のいずれかの方法で行うことができます： Flink が提供する KafkaSource ビルダークラスの setValueOnlyDeserializer(DeserializationSchema) メソッドを使用します。DeserializationSchema は、Kafka メッセージ本文のバイナリデータの解析方法を定義します。 Kafka が提供するパーサーを使用します。これには複数の実装クラスが含まれます。たとえば、StringDeserializer を使用して Kafka メッセージ本文を文字列に解析できます。 import org.apache.kafka.common.serialization.StringDeserializer; KafkaSource.<String>builder() .setDeserializer(KafkaRecordDeserializationSchema.valueOnly(StringDeserializer.class)); 説明 ConsumerRecord を完全に解析するには、KafkaRecordDeserializationSchema インターフェイスを自分で実装する必要があります。

  XML

  Kafka DataStream コネクタは、Maven Central Repository で入手できます。

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

  Kafka DataStream コネクタを使用する場合、次の Kafka プロパティを理解する必要があります：

  • 開始コンシューマーオフセット

    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))
    オフセットリセット戦略を指定せずに、コンシューマーグループによってコミットされたオフsetから消費を開始します。	KafkaSource.builder().setStartingOffsets(OffsetsInitializer.committedOffsets())

    説明

    • 組み込みのイニシャライザがニーズを満たさない場合は、カスタムオフセットイニシャライザを実装できます。

    • オフセットイニシャライザが指定されていない場合、デフォルトで OffsetsInitializer.earliest() (最も古いオフセット) が使用されます。

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

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

    説明

    通常、ストリーミングモードには停止オフセットはありません。コードのデバッグを容易にするために、setUnbounded(OffsetsInitializer) を使用してストリーミングモードで停止オフセットを指定できます。ストリーミングモードとバッチモードで停止オフセットを指定するメソッド名 (setUnbounded と setBounded) は異なることに注意してください。

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

    Flink ジョブを再起動せずにトピックのスケーリングや新しいトピックの作成などのシナリオに対応するために、提供されているトピックまたはパーティションのサブスクリプションモードで動的パーティション検出機能を有効にできます。

    説明

    動的パーティション検出機能はデフォルトで有効になっており、パーティションチェック間隔は 5 分です。この機能を無効にするには、パーティションチェック間隔を明示的に 0 以下の値に設定する必要があります。次のコードは例を示しています。

    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 ソースは、チェックポイントが完了したときに現在のコンシューマーオフセットをコミットします。これにより、Flink のチェックポイント状態が Kafka ブローカー上のコミット済みオフセットと一致することが保証されます。チェックポイントが有効になっていない場合、Kafka ソースは Kafka コンシューマーの内部自動オフセットコミットロジックに依存します。自動コミット機能は、enable.auto.commit および auto.commit.interval.ms Kafka コンシューマー設定項目によって設定されます。

    説明

    Kafka ソースは、失敗したジョブを回復するためにブローカーにコミットされたオフセットに依存しません。オフセットのコミットは、ブローカー側での Kafka コンシューマーとコンシューマーグループの消費進捗を監視するためだけです。

  • その他のプロパティ

    上記のプロパティに加えて、setProperties(Properties) および setProperty(String, String) を使用して、Kafka ソースおよび Kafka コンシューマーの任意のプロパティを設定できます。KafkaSource には通常、次の設定項目があります。

    設定項目	説明
    client.id.prefix	Kafka コンシューマーのクライアント ID プレフィックスを指定します。
    partition.discovery.interval.ms	Kafka ソースが新しいパーティションをチェックする間隔を定義します。 説明 partition.discovery.interval.ms はバッチモードでは -1 に上書きされます。
    register.consumer.metrics	Flink に Kafka コンシューマーのメトリックを登録するかどうかを指定します。
    その他の Kafka コンシューマー設定	Kafka コンシューマー設定の詳細については、「Apache Kafka」をご参照ください。 重要 Kafka コネクタは、手動で設定された一部のパラメーターを次のように強制的に上書きします： key.deserializer は常に ByteArrayDeserializer に上書きされます。 value.deserializer は常に ByteArrayDeserializer に上書きされます。 auto.offset.reset.strategy は OffsetsInitializer#getAutoOffsetResetStrategy() に上書きされます。

    次の例は、SASL メカニズムとして PLAIN を使用し、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) は、<some_parent_groups>.operator.KafkaSourceReader.topic.my-topic.partition.1.currentOffset の下に登録されます。成功したオフセットコミットの数 (commitsSucceeded) は、<some_parent_groups>.operator.KafkaSourceReader.commitsSucceeded の下に登録されます。

    • メトリックリスト

      メトリック名	説明	スコープ
      currentOffset	現在のコンシューマーオフセット。	TopicPartition
      committedOffset	現在のコミット済みオフセット。	TopicPartition
      commitsSucceeded	成功したコミット数。	KafkaSourceReader
      commitsFailed	失敗したコミット数。	KafkaSourceReader

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

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

      register.consumer.metrics 設定項目を使用して、Kafka コンシューマーのメトリックを登録するかどうかを指定できます。デフォルトでは、このオプションは true に設定されています。Kafka コンシューマーメトリックの詳細については、「Apache Kafka」をご参照ください。

• Kafka シンクの構築

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

  DataStream<String> stream = ...
  
  
  Properties properties = new Properties();
  properties.setProperty("bootstrap.servers", );
  KafkaSink<String> kafkaSink =
                  KafkaSink.<String>builder()
                          .setKafkaProducerConfig(kafkaProperties) // プロデューサー設定
                          .setRecordSerializer(
                                  KafkaRecordSerializationSchema.builder()
                                          .setTopic("my-topic") // ターゲットトピック
                                          .setKafkaValueSerializer(StringSerializer.class) // シリアル化スキーマ
                                          .build())
                          .setDeliveryGuarantee(DeliveryGuarantee.EXACTLY_ONCE) // フォールトトレランス
                          .build();
  
  stream.sinkTo(kafkaSink);

  次のパラメーターを設定する必要があります。

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

テーブル名からトピックへのマッピング戦略

Kafka をデータ統合ジョブのターゲットとして使用する場合、テーブル名からトピックへのマッピング戦略を慎重に設定する必要があります。これは、書き込まれる Kafka メッセージ形式 (Debezium JSON または Canal JSON) にもテーブル名情報が含まれており、後続の Kafka メッセージの消費では、トピック名ではなくデータ内のテーブル名が実際のテーブル名として使用されることが多いためです。

MySQL から `mydb.mytable1` と `mydb.mytable2` の 2 つのテーブルを同期する必要があるとします。考えられる設定戦略は次のとおりです：

1. マッピング戦略を設定しない

マッピング戦略がない場合、各テーブルは "database_name.table_name" 形式の名前のトピックに書き込まれます。したがって、`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. マッピングのためのルート ルールを設定する (非推奨)

多くのシナリオでは、ユーザーは書き込まれるトピックが "database_name.table_name" 形式であることを望まず、指定されたトピックにデータを書き込みたいと考えています。そのため、マッピングのためのルート ルールを設定します。次のコードは設定例を示しています：

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: mytable1

この場合、`mydb.mytable1` と `mydb.mytable2` からのすべてのデータは `mytable1` トピックに書き込まれます。

ただし、ルート ルールを使用して書き込まれるトピック名を変更すると、Kafka メッセージ (Debezium JSON または Canal JSON 形式) のテーブル名情報も変更されます。この場合、Kafka メッセージ内のすべてのテーブル名は `mytable1` になります。これにより、他のシステムがこのトピックから Kafka メッセージを消費する際に予期しない動作が発生する可能性があります。

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` からのすべてのデータは `mytable1` トピックに書き込まれます。Kafka メッセージ (Debezium JSON または Canal JSON 形式) のテーブル名情報は `mydb.mytable1` または `mydb.mytable2` のままです。他のシステムがこのトピックから Kafka メッセージを消費する際、ソーステーブル名情報を正しく取得できます。

EXACTLY_ONCE セマンティクスに関する考慮事項

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

  Kafka データを消費するすべてのアプリケーションは `isolation.level` を設定する必要があります：

  • read_committed: コミットされたデータのみを読み取ります。

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

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

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

  Flink がチェックポイントから回復する際、そのチェックポイントが開始される前にコミットされたトランザクションにのみ依存します。ジョブのクラッシュから再起動までの時間が Kafka のトランザクションタイムアウトを超えると、Kafka は自動的にトランザクションを中止し、データ損失につながります。

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

  • Flink Kafka シンクのデフォルトの transaction.timeout.ms は 1 時間です。

  • ブローカー側の transaction.max.timeout.ms を Flink の設定以上になるように増やす必要があります。

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

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

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

• 並列度のスケールインに関する制限

  最初のチェックポイントの前にジョブが失敗した場合、再起動時に元のプロデューサープール情報は保持されません。したがって、最初のチェックポイントが完了する前にジョブの並列度を下げないでください。スケールインする必要がある場合、並列度は FlinkKafkaProducer.SAFE_SCALE_DOWN_FACTOR より低くしてはなりません。

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

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

  例:

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

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

  • トランザクション 1 が完了していない限り、トランザクション 2 のデータはコンシューマーには表示されません。

  したがって:

  • 通常の操作中、データの可視性のレイテンシーはチェックポイントの間隔とほぼ同じです。

  • ジョブが失敗すると、書き込み中のトピックは、ジョブが再起動するかトランザクションがタイムアウトするまでコンシューマーをブロックします。極端な場合、トランザクションのタイムアウトが読み取りに影響を与えることさえあります。

よくある質問

• Flink は Kafka から JSON データをどのように取得しますか？

• イベント時間ベースのウィンドウを持つ Kafka ソーステーブルがデータを出力しないのはなぜですか？

• セキュリティ情報が設定された Kafka クラスターに接続するにはどうすればよいですか？

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

• ストレージ管理と操作