Realtime Compute for Apache Flink：Konektor Upsert Kafka

SQL

Konektor Upsert Kafka digunakan untuk membaca dan menulis data ke topik Kafka dengan metode upsert.

Sintaks

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

Opsi konektor dalam klausa WITH

• Opsi Umum

  Opsi	Deskripsi	Tipe data	Diperlukan	Nilai default	Catatan
  connector	Tipe tabel.	String	Ya	Tidak ada nilai default	Setel nilainya menjadi upsert-kafka.
  properties.bootstrap.servers	Alamat IP atau titik akhir dan nomor port broker Kafka.	String	Ya	Tidak ada nilai default	Format: host:port,host:port,host:port. Pisahkan beberapa pasangan host:port dengan koma (,).
  properties.*	Opsi yang dikonfigurasi untuk klien Kafka.	String	Tidak	Tidak ada nilai default	Akhiran opsi ini harus sesuai dengan aturan yang didefinisikan di Konfigurasi Produsen dan Konfigurasi Konsumen. Flink menghapus awalan properties. dan meneruskan kunci dan nilai yang telah diubah ke klien Kafka. Sebagai contoh, Anda dapat mengatur properties.allow.auto.create.topics menjadi false untuk menonaktifkan pembuatan topik otomatis. Anda tidak dapat memodifikasi konfigurasi opsi berikut dengan menambahkan awalan properties., karena nilai opsi tersebut ditimpa setelah Anda menggunakan konektor Upsert Kafka: key.deserializer value.deserializer
  key.format	Format yang digunakan untuk membaca atau menulis bidang kunci pesan Kafka.	String	Ya	Tidak ada nilai default	Anda harus mengonfigurasi opsi key.fields atau key.fields-prefix jika Anda mengonfigurasi opsi ini. Nilai 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. Anda dapat mengonfigurasi parameter ini untuk mencegah konflik nama dengan bidang nilai.	String	Tidak	Tidak ada nilai default	Opsi ini hanya digunakan untuk membedakan nama kolom tabel sumber dan tabel sink. Awalan dihapus dari nama kolom saat bidang kunci pesan Kafka diurai dan dibuat. Catatan Jika Anda mengonfigurasi opsi ini, Anda harus mengatur opsi value.fields-include menjadi EXCEPT_KEY.
  value.format	Format yang digunakan untuk membaca atau menulis bidang nilai pesan Kafka.	String	Ya	Tidak ada nilai default	Konfigurasi opsi ini setara dengan konfigurasi opsi format. Opsi format tidak dapat digunakan bersama dengan opsi value.format. Jika Anda mengonfigurasi kedua opsi tersebut, terjadi konflik.
  value.fields-include	Menentukan apakah akan menyertakan bidang yang sesuai dengan kunci pesan saat bidang nilai pesan Kafka diurai atau dibuat.	String	Ya	ALL	Nilai valid: ALL: Semua bidang diproses sebagai bidang nilai pesan Kafka. Ini adalah nilai default. EXCEPT_KEY: Semua bidang kecuali bidang yang ditentukan oleh opsi key.fields diproses sebagai bidang nilai pesan Kafka.
  topic	Nama topik tempat data dibaca atau ke mana data ditulis.	String	Ya	Tidak ada nilai default	N/A

• Opsi Eksklusif Sink

  Opsi	Deskripsi	Tipe data	Diperlukan	Nilai default	Catatan
  sink.parallelism	Paralelisme operator dalam tabel sink Kafka.	Integer	Tidak	Paralelisme operator hulu, yang ditentukan oleh kerangka kerja.	N/A
  sink.buffer-flush.max-rows	Jumlah maksimum catatan data yang dapat di-cache sebelum cache diperbarui.	Integer	Tidak	0 (dinonaktifkan)	Jika tabel sink menerima sejumlah besar pembaruan pada kunci yang sama, hanya catatan data terakhir dari kunci tersebut yang disimpan dalam cache. Dalam hal ini, caching data dalam tabel sink membantu mengurangi jumlah data yang ditulis ke topik Kafka. Ini mencegah pesan nisan potensial dikirim ke topik Kafka. Catatan Jika Anda ingin mengaktifkan caching data untuk tabel sink, Anda harus mengatur opsi sink.buffer-flush.max-rows dan sink.buffer-flush.interval ke nilai yang lebih besar dari 0.
  sink.buffer-flush.interval	Interval pembaruan cache.	Durasi	Tidak	0 (dinonaktifkan)	Unit bisa milidetik, detik, menit, atau jam. Sebagai contoh, Anda dapat mengonfigurasi 'sink.buffer-flush.interval'='1 s'. Jika tabel sink menerima sejumlah besar pembaruan pada kunci yang sama, hanya catatan data terakhir dari kunci tersebut yang disimpan dalam cache. Dalam hal ini, caching data dalam tabel sink membantu mengurangi jumlah data yang ditulis ke topik Kafka. Ini mencegah pesan nisan potensial dikirim ke topik Kafka. Catatan Jika Anda ingin mengaktifkan caching data untuk tabel sink, Anda harus mengatur opsi sink.buffer-flush.max-rows dan sink.buffer-flush.interval ke nilai yang lebih besar dari 0.

Ingest data

Konektor Upsert Kafka dapat digunakan untuk mengembangkan draft YAML untuk ingest data dan sebagai sink untuk menulis data dalam format JSON, termasuk bidang kunci utama dalam tubuh pesan.

Sintaks

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}

Opsi konektor

Tipe perubahan yang didukung

Konektor Upsert Kafka mendukung semua tipe perubahan untuk pemasukan data. Namun, perlu diperhatikan bahwa data harus dimasukkan dari tabel sumber dengan skema tetap melalui pekerjaan Flink SQL menggunakan Konektor Upsert Kafka.