Realtime Compute for Apache Flink:アップサート Kafka

SQL

アップサート Kafka コネクタは、アップサート操作を使用して Kafka トピックとの間でデータを読み書きします。

構文

CREATE TABLE upsert_kafka_sink(
user_region STRING,
pv BIGINT,
uv BIGINT,
PRIMARY KEY(user_region) NOT ENFORCED
)WITH(
'connector'='upsert-kafka',
'topic'='<yourTopicName>',
'properties.bootstrap.servers'='...',
'key.format'='avro',
'value.format'='avro'
);

WITH パラメーター

• 一般

  パラメーター	説明	データ型	必須	デフォルト	備考
  connector	テーブルのタイプ。	String	はい	なし	値を `upsert-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
  key.format	Kafka メッセージのキー部分のフォーマット。	String	はい	なし	このパラメーターを構成する場合は、key.fields または key.fields-prefix も構成する必要があります。 有効な値: csv json avro debezium-json canal-json maxwell-json avro-confluent raw
  key.fields-prefix	Kafka メッセージ内のすべてのキーフィールドのカスタムプレフィックス。これにより、値フォーマット内のフィールドとの名前衝突を回避します。	String	いいえ	なし	このパラメーターは、ソーステーブルとシンクテーブル間の列名を区別するためにのみ使用されます。Kafka メッセージのキー部分が解析および生成されるときに、プレフィックスは削除されます。 説明 このパラメーターを構成する場合は、value.fields-include を `EXCEPT_KEY` に設定する必要があります。
  value.format	Kafka メッセージの値部分のフォーマット。	String	はい	なし	このパラメーターは format と同等です。どちらか一方のみを構成できます。format と value.format の両方を構成すると競合が発生します。
  value.fields-include	Kafka メッセージの値部分が解析または生成されるときに、メッセージキーに対応するフィールドを含めるかどうかを指定します。	String	はい	ALL	有効な値: ALL (デフォルト): すべての列が Kafka メッセージの値部分として処理されます。 EXCEPT_KEY: `key.fields` で定義されたフィールドを除くすべての列が Kafka メッセージの値部分として処理されます。
  topic	読み取りまたは書き込みを行うトピック名。	String	はい	なし	なし。

• シンク固有のパラメーター

  パラメーター	説明	データ型	必須	デフォルト	備考
  sink.parallelism	Kafka シンク演算子の同時実行数。	Integer	いいえ	アップストリーム演算子の同時実行数。これはフレームワークによって決定されます。	なし。
  sink.buffer-flush.max-rows	キャッシュがフラッシュされる前にキャッシュできる最大レコード数。	Integer	いいえ	0 (無効)	シンクテーブルが同じキーに対して多くの更新を受信する場合、キャッシュはそのキーの最後のレコードのみを保持します。これにより、Kafka トピックに送信されるデータ量を削減し、潜在的なトムストーンメッセージを回避できます。 説明 シンクキャッシュを有効にするには、sink.buffer-flush.max-rows と sink.buffer-flush.interval の両方をゼロより大きい値に設定します。
  sink.buffer-flush.interval	キャッシュがフラッシュされる間隔。	Duration	いいえ	0 (無効)	単位はミリ秒 (ms)、秒 (s)、分 (min)、または時間 (h) です。例: 'sink.buffer-flush.interval'='1 s'。 シンクテーブルが同じキーに対して多くの更新を受信する場合、キャッシュはそのキーの最後のレコードのみを保持します。これにより、Kafka トピックに送信されるデータ量を削減し、潜在的なトムストーンメッセージを回避できます。 説明 シンクキャッシュを有効にするには、sink.buffer-flush.max-rows と sink.buffer-flush.interval の両方をゼロより大きい値に設定します。

データインジェスト

アップサート Kafka コネクタは、YAML データインジェストジョブのシンクとして使用できます。データは JSON フォーマットで書き込まれ、プライマリキーフィールドもメッセージ本文に含まれます。

構文

sink:
  type: upsert-kafka
  name: upsert-kafka Sink
  properties.bootstrap.servers: localhost:9092
  # ApsaraMQ for Kafka
  aliyun.kafka.accessKeyId: ${secret_values.kafka-ak}
  aliyun.kafka.accessKeySecret: ${secret_values.kafka-sk}
  aliyun.kafka.instanceId: ${instancd-id}
  aliyun.kafka.endpoint: ${endpoint}
  aliyun.kafka.regionId: ${region-id}

パラメーター

サポートされているタイプ変更

データインジェスト用のアップサート Kafka コネクタは、すべてのタイプの変更操作をサポートしています。ただし、データを読み取るには、固定スキーマを持つ Flink アップサート Kafka SQL コネクタを使用する必要があります。