Realtime Compute for Apache Flink:Upsert Kafka

SQL

Konektor Upsert Kafka membaca data dari dan menulis data ke topik Kafka menggunakan operasi upsert.

Sintaksis

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

DENGAN parameter

• Umum

  Parameter	Deskripsi	Tipe data	Wajib	Bawaan	Keterangan
  connector	Jenis tabel.	String	Ya	None	Atur nilainya ke `upsert-kafka`.
  properties.bootstrap.servers	Alamat broker Kafka.	String	Ya	None	Formatnya adalah host:port,host:port,host:port. Pisahkan beberapa alamat dengan koma (,).
  properties.*	Parameter untuk client Kafka.	String	Tidak	None	Akhiran harus merupakan konfigurasi yang didefinisikan dalam dokumentasi resmi Kafka untuk producer dan consumer. Flink menghapus awalan `properties.` dan meneruskan konfigurasi yang tersisa ke client Kafka. Misalnya, Anda dapat menggunakan 'properties.allow.auto.create.topics' = 'false' untuk menonaktifkan pembuatan topik otomatis. Jangan ubah konfigurasi berikut dengan cara ini, karena konektor Kafka akan menimpanya: key.deserializer value.deserializer
  key.format	Format untuk bagian kunci pesan Kafka.	String	Ya	None	Jika Anda mengonfigurasi parameter ini, Anda juga harus mengonfigurasi key.fields atau key.fields-prefix. Nilai yang valid: csv json avro debezium-json canal-json maxwell-json avro-confluent raw
  key.fields-prefix	Awalan kustom untuk semua bidang kunci dalam pesan Kafka. Hal ini menghindari konflik nama dengan bidang dalam format nilai.	String	Tidak	None	Parameter ini hanya digunakan untuk membedakan nama kolom antara tabel sumber dan tabel sink. Awalan dihapus saat bagian kunci pesan Kafka diurai dan dihasilkan. Catatan Jika Anda mengonfigurasi parameter ini, Anda harus mengatur value.fields-include ke `EXCEPT_KEY`.
  value.format	Format untuk bagian nilai pesan Kafka.	String	Ya	None	Parameter ini setara dengan format. Anda hanya dapat mengonfigurasi salah satunya. Konflik terjadi jika Anda mengonfigurasi kedua parameter format dan value.format.
  value.fields-include	Menentukan apakah bidang yang sesuai dengan kunci pesan disertakan saat bagian nilai pesan Kafka diurai atau dihasilkan.	String	Ya	ALL	Nilai yang valid: ALL (Default): Semua kolom diproses sebagai bagian nilai pesan Kafka. EXCEPT_KEY: Semua kolom kecuali bidang yang didefinisikan oleh `key.fields` diproses sebagai bagian nilai pesan Kafka.
  topic	Nama topik untuk dibaca atau ditulis.	String	Ya	None	None.

• Parameter khusus sink

  Parameter	Deskripsi	Tipe data	Wajib	Bawaan	Keterangan
  sink.parallelism	Konkurensi operator sink Kafka.	Integer	Tidak	Konkurensi operator hulu, yang ditentukan oleh framework.	None.
  sink.buffer-flush.max-rows	Jumlah maksimum catatan yang dapat di-cache sebelum cache dikosongkan.	Integer	Tidak	0 (dinonaktifkan)	Jika tabel sink menerima banyak pembaruan pada kunci yang sama, cache hanya menyimpan catatan terakhir untuk kunci tersebut. Hal ini membantu mengurangi jumlah data yang dikirim ke topik Kafka dan menghindari potensi pesan tombstone. Catatan Untuk mengaktifkan caching sink, atur kedua parameter sink.buffer-flush.max-rows dan sink.buffer-flush.interval ke nilai yang lebih besar dari nol.
  sink.buffer-flush.interval	Interval pengosongan cache.	Duration	Tidak	0 (dinonaktifkan)	Unitnya dapat berupa milidetik (ms), detik (s), menit (min), atau jam (h). Contohnya, 'sink.buffer-flush.interval'='1 s'. Jika tabel sink menerima banyak pembaruan pada kunci yang sama, cache hanya menyimpan catatan terakhir untuk kunci tersebut. Hal ini membantu mengurangi jumlah data yang dikirim ke topik Kafka dan menghindari potensi pesan tombstone. Catatan Untuk mengaktifkan caching sink, atur kedua parameter sink.buffer-flush.max-rows dan sink.buffer-flush.interval ke nilai yang lebih besar dari nol.

Ingesti data

Konektor Upsert Kafka dapat digunakan sebagai sink dalam pekerjaan ingesti data YAML. Data ditulis dalam format JSON, dan bidang kunci primer juga disertakan dalam isi pesan.

Sintaksis

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}

Parameter

Perubahan tipe yang didukung

Konektor Upsert Kafka untuk ingesti data mendukung semua jenis operasi perubahan. Namun, untuk membaca data tersebut, Anda harus menggunakan konektor SQL Flink Upsert Kafka dengan skema tetap.