Realtime Compute for Apache Flink:Upsert Kafka コネクタ

SQL

Upsert 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ブローカーの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 に設定して、トピックの自動作成を無効にすることができます。 Upsert Kafkaコネクタを使用した後、パラメーターの値が上書きされるため、properties.プレフィックスを追加して次のパラメーターの構成を変更することはできません。 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トピックに書き込まれるデータ量を削減するのに役立ちます。これにより、潜在的な墓石メッセージがKafkaトピックに送信されるのを防ぎます。 説明 シンクテーブルのデータキャッシュを有効にするには、sink.buffer-flush.max-rowsおよびsink.buffer-flush.intervalパラメーターを 0より大きい値に設定する必要があります。
  sink.buffer-flush.interval	キャッシュが更新される間隔。	Duration	いいえ	0（無効）	単位はミリ秒、秒、分、または時間です。 たとえば、'sink.buffer-flush.interval'='1 s' と構成できます。 シンクテーブルが同じキーに対して多数の更新を受信した場合、キーの最後のデータレコードのみがキャッシュに保持されます。この場合、シンクテーブルでのデータキャッシュは、Kafkaトピックに書き込まれるデータ量を削減するのに役立ちます。これにより、潜在的な墓石メッセージがKafkaトピックに送信されるのを防ぎます。 説明 シンクテーブルのデータキャッシュを有効にするには、sink.buffer-flush.max-rowsおよびsink.buffer-flush.intervalパラメーターを 0より大きい値に設定する必要があります。

データインジェスチョン

Upsert 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}

コネクタオプション

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

Upsert Kafka コネクタは、データ ingestion のすべての変更タイプをサポートしています。 ただし、Upsert Kafka コネクタを使用して Flink SQL ジョブを介して固定スキーマを持つソーステーブルからデータを ingestion する必要があることに注意してください。