Realtime Compute for Apache Flink:Kafka コネクタ

概要

前提条件

注意事項

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

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

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

• 複数の Apache Flink ジョブが同じ sink.transactional-id-prefix を使用すると、生成されるトランザクション ID が競合する可能性があります。あるジョブで書き込み操作が失敗すると、Apache Kafka パーティションの Log Start Offset (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 ドキュメントの 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	いいえ	–	例： user_event_.*：user_event_ で始まるすべてのトピックに一致します。 prod\.logs\..*：prod.logs. で始まるトピックに一致します (. 文字はエスケープする必要があります)。 説明 このオプションまたは topic のいずれかを指定できますが、両方は指定できません。
  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 のコミット済みオフセットから読み取りを開始します。 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 を使用する場合、コネクタはパターンに一致する新しいトピックも検出します。この機能を無効にするには、間隔を正でない値に設定します。 説明 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 データを保持します。 説明 このオプションは、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	並列シンクインスタンスからのレコードを Kafka パーティションにマッピングします。	String	いいえ	default	有効な値： default：デフォルトの Kafka パーティショナーを使用します。 fixed：各並列シンクインスタンスは固定の Kafka パーティションに書き込みます。 round-robin：レコードはラウンドロビン方式でパーティションに分散されます。 カスタムパーティショナー：カスタムパーティショナーを使用するには、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) からファイルを動的にダウンロードすることを回避しますが、デバッグモードはサポートされません。

• 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 の組み込みパーティショニング戦略が要件を満たさない場合は、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 への 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'
);

1回限りのセマンティクス

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

  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 クラスターに接続するにはどうすればよいですか？

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

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