Realtime Compute for Apache Flink:Message Queue for Kafka

背景情報

前提条件

注意事項

現在、Flink および Kafka コミュニティの設計上の制限により、トランザクション書き込みは推奨されません。sink.delivery-guarantee = exactly-once を設定すると、Kafka コネクタはトランザクション書き込みを有効にしますが、これには 3 つの既知の問題があります：

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

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

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

exactly-once セマンティクスが必要な場合は、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 ブローカーの IP アドレスとポート番号。	String	はい	なし	フォーマット：host:port,host:port,host:port。複数の host:port ペアはカンマ (,) で区切ります。
  properties.*	Kafka クライアント用に設定されるオプション。	String	いいえ	なし	サフィックスは、Producer Configs および Consumer Configs で定義されているルールに準拠する必要があります。 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-{ソーステーブル名}	指定されたグループ 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 オプションで指定された 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	Kafka のトピックとパーティションを動的に検出する時間間隔。	Duration	いいえ	5 分	デフォルトのパーティション検出間隔は 5 分です。この機能を無効にするには、このオプションを明示的に非正の値に設定する必要があります。動的パーティション検出機能が有効になると、Kafka ソースは新しいパーティションを自動的に検出し、そのパーティションからデータを読み取ることができます。topic-pattern モードでは、Kafka ソースは既存のトピックの新しいパーティションからデータを読み取り、正規表現に一致する新しいトピックのすべてのパーティションからデータを読み取ります。 説明 VVR 6.0.X を使用する Realtime Compute for Apache Flink では、動的パーティション検出機能はデフォルトで無効になっています。VVR 8.0 以降を使用する Realtime Compute for Apache Flink では、この機能はデフォルトで有効になっています。デフォルトのパーティション検出間隔は 5 分です。
  scan.header-filter	データが特定のメッセージヘッダーを含むかどうかに基づく Kafka データフィルタリング。	String	いいえ	なし	ヘッダーキーと値はコロン (:) で区切ります。複数のヘッダーは AND (&) や OR (|) などの論理演算子で区切ります。論理演算子 NOT (!) がサポートされています。たとえば、depart:toy|depart:book&!env:test は、ヘッダーに depart=toy または depart=book を含み、かつ env=test を含まない Kafka データを保持することを示します。 説明 このオプションは、VVR 8.0.6 以降を使用する Realtime Compute for Apache Flink でのみサポートされます。 括弧操作はサポートされていません。 論理演算は左から右へ順に実行されます。 UTF-8 形式のヘッダー値は文字列に変換され、scan.header-filter オプションで指定されたヘッダー値と比較されます。
  scan.check.duplicated.group.id	properties.group.id パラメーターで指定された重複するコンシューマーグループをチェックするかどうかを指定します。	Boolean	いいえ	false	有効な値： true：ジョブが開始する前に重複するコンシューマーグループをチェックします。重複するコンシューマーグループが存在する場合、エラーを報告し、競合を防ぐためにジョブを一時停止します。 false：ジョブが開始する前に重複するコンシューマーグループをチェックしません。 説明 このオプションは、VVR 6.0.4 以降を使用する Realtime Compute for Apache Flink でのみサポートされます。

• シンク固有

  パラメーター	説明	データの型	必須	デフォルト値	備考
  topic	データが書き込まれるトピックの名前。	String	はい	なし	N/A
  sink.partitioner	Flink の並列度を Kafka のパーティションにマッピングするパターン。	String	いいえ	default	有効な値： default (デフォルト)：デフォルトの Kafka パーティショナーを使用してデータをパーティション分割します。 fixed：各 Flink パーティションは固定の Kafka パーティションに対応します。 round-robin：Flink パーティション内のデータは、ラウンドロビンシーケンスで Kafka パーティションに分散されます。 カスタムパーティションマッピングパターン：FlinkKafkaPartitioner のサブクラスを作成して、カスタムパーティションマッピングパターンを設定できます。例：org.mycompany.MyPartitioner。
  sink.delivery-guarantee	Kafka シンクテーブルのセマンティックパターン。	String	いいえ	at-least-once	有効な値： none：配信セマンティクスは保証されません。データが失われたり、重複したりする可能性があります。 at-least-once (デフォルト)：データが失われないことを保証します。ただし、データが重複する可能性があります。 exactly-once：Kafka トランザクションを使用して exactly-once セマンティクスを保証します。これにより、データが失われたり、重複したりしないことが保証されます。 説明 このオプションを 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',
  /*Secure Sockets Layer (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',
  /*クライアントがサーバーアドレスを検証するために使用するアルゴリズム。null 値はサーバーアドレス検証が無効であることを示します。*/
  '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 証明書と秘密鍵をアップロードできます。アップロード後、ファイルは /flink/usrlib ディレクトリに保存されます。使用したい CA 証明書ファイルの名前が my-truststore.jks の場合、WITH 句の 'properties.ssl.truststore.location' パラメーターを次の 2 つの方法で設定して、この証明書を使用できます：

• 'properties.ssl.truststore.location' = '/flink/usrlib/my-truststore.jks' を設定した場合、Flink は実行時に OSS ファイルを動的にダウンロードする必要はありませんが、デバッグモードはサポートされません。

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

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

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 でのデータ読み取りの進行状況を監視するためにのみ使用されます。オフセットのコミットに失敗しても、データの正確性には影響しません。

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

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

Kafka、Upsert Kafka、Kafka JSON カタログの比較

Kafka はデータ挿入のみをサポートし、更新や削除をサポートしないメッセージキューシステムです。そのため、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>',
  '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 ソースは、KafkaSource のインスタンスを作成するためのビルダークラスを提供します。次のサンプルコードは、"input-topic" トピックの最も古いオフセットからメッセージを消費し、コンシューマーグループ名を my-group とし、Kafka メッセージ本文を文字列として逆シリアル化する 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) メソッドを呼び出して ID を設定できます。
  Topics または Partition	サブスクライブするトピックまたはパーティションの名前。以下のサブスクリプションパターンのいずれかを使用して、トピックまたはパーティションにサブスクライブするように Kafka ソースを設定できます： トピックリスト。トピックリストを設定すると、Kafka ソースは指定されたトピックのすべてのパーティションにサブスクライブします。 KafkaSource.builder().setTopics("topic-a","topic-b") トピックパターン。正規表現を指定すると、Kafka ソースは指定された正規表現に一致するトピックのすべてのパーティションにサブスクライブします。 KafkaSource.builder().setTopicPattern("topic.*") パーティションリスト。パーティションリストを設定すると、Kafka ソースは指定されたパーティションにサブスクライブします。 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)
  Deserializer	Kafka メッセージを逆シリアル化するデシリアライザ。 setDeserializer(KafkaRecordDeserializationSchema) メソッドを呼び出してデシリアライザを指定できます。KafkaRecordDeserializationSchema インターフェイスは、ConsumerRecord オブジェクトがどのように逆シリアル化されるかを定義します。ConsumerRecord オブジェクトの Kafka メッセージの Value フィールドのみを逆シリアル化するには、以下のいずれかの方法を使用できます： Kafka ソースは setValueOnlyDeserializer(DeserializationSchema) メソッドを提供します。DeserializationSchema クラスは、バイナリ値として保存されている Kafka メッセージがどのように逆シリアル化されるかを定義します。 Kafka の Deserializer インターフェイスを実装するクラスを使用します。たとえば、StringDeserializer クラスを使用してメッセージを文字列に逆シリアル化できます。 import org.apache.kafka.common.serialization.StringDeserializer; KafkaSource.<String>builder() .setDeserializer(KafkaRecordDeserializationSchema.valueOnly(StringDeserializer.class)); 説明 ConsumerRecord オブジェクトを逆シリアル化したい場合は、KafkaRecordDeserializationSchema インターフェイスを実装するクラスを作成する必要があります。

  XML

  Kafka DataStream コネクタは Maven 中央リポジトリに保存されています。

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

  Kafka DataStream コネクタを使用する際には、以下の Kafka プロパティに精通している必要があります：

  • 開始オフセット

    Kafka ソースがデータの読み取りを開始する際に、オフセットイニシャライザを使用してオフセットを指定できます。オフセットイニシャライザは、OffsetsInitializer インターフェイスを実装するオブジェクトです。KafkaSource クラスは、以下の組み込みオフセットイニシャライザを提供します。

    オフセットイニシャライザ	コード設定
    最も古いオフセットからデータを読み取ります。	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 ソースはストリーミングモードで動作します。このモードでは、デプロイメントは失敗するかキャンセルされるまで実行し続けます。Kafka ソースをバッチモードで動作させたい場合は、setBounded(OffsetsInitializer) メソッドを呼び出して停止オフセットを指定できます。すべてのパーティションが停止オフセットに達すると、Kafka ソースは終了します。

    説明

    ほとんどの場合、ストリーミングモードで動作する Kafka ソースには停止オフセットがありません。ストリーミングモードで動作する Kafka ソースをデバッグしたい場合は、setUnbounded(OffsetsInitializer) メソッドを呼び出して停止オフセットを指定できます。停止オフセットを指定する方法は、ストリーミングモードとバッチモードのどちらを使用するかによって異なります。

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

    実行中のデプロイメントが、デプロイメントを再起動することなく、サブスクリプションパターンに一致する新しいトピックや新しいパーティションからのデータを処理するようにしたい場合は、Kafka ソースで動的パーティション検出機能を有効にできます。DataStream コネクタでは、この機能はデフォルトで無効になっており、手動で有効にする必要があります：

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

    重要

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

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

    デフォルトでは、Kafka ソースはレコードに添付されたタイムスタンプをそのレコードのイベント時間として使用します。各レコードのイベント時間に基づいてウォーターマーク戦略を定義し、ウォーターマークを下流のサービスに送信できます。

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

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

    説明

    一部のソースサブタスクが長期間アイドル状態になると (たとえば、Kafka パーティションが新しいメッセージを受信しない、またはソースの並列度が Kafka パーティション数を超えるなど)、ウォーターマークの生成が失敗する可能性があります。この場合、ウィンドウ計算がトリガーされず、データ処理が停止します。

    解決策は次のとおりです：

    • ウォーターマークのタイムアウトメカニズムを設定する：table.exec.source.idle-timeout パラメーターを有効にして、指定されたタイムアウト期間後にシステムがウォーターマークを強制的に生成するようにし、ウィンドウ計算エポックの進行を確保します。

    • データソースを最適化する：ソースの並列度を Kafka パーティション数以下に設定します。

  • コンシューマーオフセットのコミット

    チェックポイントが生成されると、Kafka ソースは各パーティションの Kafka コンシューマーオフセットを Kafka ブローカーにコミットします。これにより、Kafka ブローカーに記録された Kafka コンシューマーオフセットがチェックポイントの状態と一致することが保証されます。Kafka コンシューマーは、各パーティションのオフセットを定期的に Kafka ブローカーに自動的にコミットできます。自動オフセットコミット機能は、enable.auto.commit および auto.commit.interval.ms オプションを使用して設定できます。チェックポイント機能を無効にすると、Kafka ソースは Kafka コンシューマーに依存してオフセットを Kafka ブローカーにコミットします。

    説明

    Kafka ソースは、フォールトトレランスのために Kafka ブローカーに記録されたコミット済みオフセットを使用しません。オフセットをコミットすると、Kafka ブローカーは各パーティションでのレコード消費の進行状況を監視できます。

  • 追加のプロパティ

    setProperties(Properties) および setProperty(String, String) メソッドを呼び出して、Kafka ソースおよび Kafka コンシューマーに追加のプロパティを設定できます。次の表は、Kafka ソースのプロパティについて説明します。

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

    次のサンプルコードは、Kafka コンシューマーが JAAS 設定と SASL/PLAIN 認証メカニズムを使用して Kafka クラスターに接続する方法を示しています。

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

  • 監視

    Kafka ソースは、監視と診断のために Realtime Compute for Apache Flink にメトリックを登録します。

    • メトリクススコープ

      Kafka ソースのすべてのメトリックは、KafkaSourceReader メトリックグループの下に登録されます。KafkaSourceReader は、オペレーターメトリックグループのサブグループです。特定のパーティションのメトリックは、KafkaSourceReader.topic.<topic_name>.partition.<partition_id> メトリックグループに登録されます。

      たとえば、トピック名が my-topic で、トピックのパーティション名が 1 の場合、パーティションのコンシューマーオフセットは <some_parent_groups>.operator.KafkaSourceReader.topic.my-topic.partition.1.currentOffset メトリックによって報告されます。コンシューマーオフセットの成功したコミット数は、<some_parent_groups>.operator.KafkaSourceReader.commitsSucceeded メトリックによって測定されます。

    • メトリック

      メトリック	説明	範囲
      currentOffset	現在のコンシューマーオフセット	TopicPartition
      committedOffset	現在のコミットオフセット	TopicPartition
      commitsSucceeded	成功した送信数	KafkaSourceReader
      commitsFailed	失敗した送信数	KafkaSourceReader

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

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

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

• Kafka シンクの作成

  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);

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

  パラメーター	説明
  Topic	データが書き込まれるトピックの名前。
  データシリアル化	Kafka シンクを構築する際には、入力データを Kafka の ProducerRecord オブジェクトに変換するために KafkaRecordSerializationSchema を提供する必要があります。Flink は、メッセージキーと値のシリアル化、トピックの選択、メッセージのパーティショニングなどの一般的なコンポーネントを提供するスキーマビルダーを提供します。より高度な制御のために、対応するインターフェイスを実装することもできます。Kafka シンクは、受信する各レコードに対して ProducerRecord<byte[], byte[]> serialize(T element, KafkaSinkContext context, Long timestamp) メソッドを呼び出し、シリアル化されたレコードを表す ProducerRecord オブジェクトを生成します。その後、Kafka シンクはその ProducerRecord オブジェクトを必要なトピックに書き込みます。 各レコードが Kafka にどのように書き込まれるかを詳細に制御できます。ProducerRecord を使用して、以下の操作を実行できます： ターゲットトピックの名前を設定します。 メッセージキーを定義します。 ターゲットパーティションを指定します。
  Kafka クライアントプロパティ	bootstrap.servers プロパティは必須です。カンマ区切りの Kafka ブローカーアドレスのリストを指定します。
  フォールトトレランスセマンティクス	チェックポイント機能を有効にすると、Kafka シンクは exactly-once 配信を保証できます。また、DeliveryGuarantee パラメーターを設定して、異なるフォールトトレランスセマンティクスを指定することもできます。DeliveryGuarantee パラメーターの詳細は以下のとおりです： DeliveryGuarantee.NONE：Flink による配信保証は提供されません。データが失われたり、重複したりする可能性があります。 DeliveryGuarantee.AT_LEAST_ONCE：Kafka シンクはデータが失われないことを保証します。ただし、データが重複する可能性があります。 DeliveryGuarantee.EXACTLY_ONCE：Kafka シンクはデータが失われたり、重複したりしないことを保証します。Kafka トランザクションメカニズムを使用して exactly-once 配信を保証します。 説明 exactly-once セマンティクスの詳細については、「Semantic.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
  # 最大 1000 件のダーティデータレコードを許容します。
  ingestion.error-tolerance.max-count: 1000

この設定により、最大 1000 件のダーティデータレコードが許容され、少量のダーティデータが存在する場合でもジョブが正常に実行されます。ダーティデータレコードの数がこのしきい値を超えると、ジョブは失敗し、データの検証が必要になります。

ダーティデータによってジョブが失敗しないことを保証するために、次の設定を使用できます：

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

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

Kafka をデータインジェストシンクとして使用する場合、メッセージ形式 (debezium-json や canal-json) には多くの場合テーブル名情報が含まれています。コンシューマーは通常、トピック名ではなく、このテーブル名を実際のテーブル名として使用します。したがって、テーブル名とトピックのマッピング戦略を注意深く設定する必要があります。

MySQL から mydb.mytable1 および mydb.mytable2 の 2 つのテーブルを同期する必要があると仮定します。可能なマッピング戦略は以下のとおりです：

1. 何のマッピング戦略も設定しない

マッピング戦略を設定しない場合、各テーブルはデータベース名とテーブル名を基にした名前のトピック (例：mydb.mytable1) に書き込まれます。したがって、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: mytable1

この場合、mydb.mytable1 および mydb.mytable2 のすべてのデータが mytable1 トピックに書き込まれます。

ただし、ルートルールを使用してトピック名を変更すると、Kafka メッセージ (debezium-json や canal-json 形式) 内のテーブル名も変更されます。その結果、このトピック内のすべてのメッセージのテーブル名は mytable1 になります。このトピックからメッセージを消費する他のシステムは、予期しない動作をする可能性があります。

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` トピックに書き込まれますが、元のテーブル名 (`mydb.mytable1` または `mydb.mytable2`) は Kafka メッセージ形式 (`debezium-json` や `canal-json`) 内で保持されます。これにより、このトピックからメッセージを消費する他のシステムが、ソーステーブル名情報を正しく取得できます。

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 の設定以上に増やす必要があります。

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

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

  プールサイズは、最大同時チェックポイント数に応じて調整してください。

• スケールダウンの制限

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

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

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

  たとえば：

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

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

  • トランザクション 1 が完了するまで、トランザクション 2 のデータはコンシューマーには表示されません。

  したがって：

  • 通常の運用では、データの可視性は平均チェックポイント間隔分遅延します。

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

よくある質問

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

• イベント時間ベースのウィンドウを使用して Kafka ソーステーブルのデータを処理した後、データ出力がないのはなぜですか？

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

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

• ストレージ管理および操作