Realtime Compute for Apache Flink:Kafka connector

Ikhtisar

Prasyarat

Catatan

Penulisan transaksional tidak disarankan karena adanya keterbatasan desain yang diketahui pada Apache Flink dan Apache Kafka. Saat Anda mengatur sink.delivery-guarantee = 'exactly-once', Kafka connector mengaktifkan penulisan transaksional, dengan masalah yang diketahui sebagai berikut:

• Setiap checkpoint menghasilkan ID Transaksi baru. Jika interval checkpoint terlalu pendek, banjir ID Transaksi yang dihasilkan dapat menyebabkan koordinator kluster Kafka kehabisan memori, sehingga mengganggu stabilitas kluster.

• Setiap transaksi membuat instans Producer baru. Jika terlalu banyak transaksi melakukan commit secara bersamaan, Pengelola Tugas (TaskManager) dapat kehabisan memori, sehingga mengganggu pekerjaan Apache Flink.

• Jika beberapa pekerjaan Apache Flink menggunakan sink.transactional-id-prefix yang sama, ID Transaksi yang dihasilkan dapat bertabrakan. Ketika operasi penulisan gagal pada satu pekerjaan, hal ini dapat mencegah Log Start Offset (LSO) partisi Apache Kafka maju. Hal ini memengaruhi semua konsumen partisi tersebut.

Jika Anda memerlukan semantik tepat-sekali, gunakan connector Upsert Kafka untuk menulis ke tabel kunci primer, sehingga memastikan idempotensi. Jika Anda harus menggunakan penulisan transaksional, lihat Catatan penggunaan semantik tepat-sekali.

Pemecahan masalah konektivitas jaringan

Error Timed out waiting for a node assignment saat pekerjaan Realtime Compute for Apache Flink gagal dimulai biasanya menunjukkan adanya masalah konektivitas jaringan antara Realtime Compute for Apache Flink dan kluster Kafka.

SQL

Gunakan Kafka connector sebagai tabel source atau tabel sink dalam pekerjaan SQL.

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

Kolom metadata

CREATE TABLE kafka_source (
  -- Baca topik pesan sebagai kolom `record_topic`
  `record_topic` STRING NOT NULL METADATA FROM 'topic' VIRTUAL,
  -- Baca timestamp dari ConsumerRecord sebagai kolom `ts`
  `ts` TIMESTAMP_LTZ(3) METADATA FROM 'timestamp' VIRTUAL,
  -- Baca offset pesan sebagai kolom `record_offset`
  `record_offset` BIGINT NOT NULL METADATA FROM 'offset' VIRTUAL,
  ...
) WITH (
  'connector' = 'kafka',
  ...
);

CREATE TABLE kafka_sink (
  -- Tulis timestamp dari kolom `ts` sebagai timestamp ProducerRecord ke Kafka
  `ts` TIMESTAMP_LTZ(3) METADATA FROM 'timestamp' VIRTUAL,
  ...
) WITH (
  'connector' = 'kafka',
  ...
);

Opsi connector

• Umum

  Opsi	Deskripsi	Jenis	Wajib	Default	Keterangan
  connector	Jenis connector.	String	Ya	–	Nilainya harus kafka.
  properties.bootstrap.servers	Daftar alamat broker Kafka.	String	Ya	–	Format: host:port,host:port,.... Pisahkan alamat dengan koma (,).
  properties.*	Properti tambahan untuk klien Kafka.	String	Tidak	–	Kunci properti harus merupakan opsi yang valid sebagaimana didefinisikan dalam dokumentasi resmi Apache Kafka untuk Producer Configs dan Consumer Configs. Realtime Compute for Apache Flink menghapus awalan properties. dan meneruskan pasangan kunci-nilai yang tersisa ke klien Kafka yang mendasari. Misalnya, Anda dapat mengatur 'properties.allow.auto.create.topics' = 'false' untuk menonaktifkan pembuatan topik otomatis. Kafka connector menimpa opsi-opsi ini, sehingga Anda tidak dapat mengonfigurasinya dengan cara ini: key.deserializer value.deserializer
  format	Format untuk serialisasi dan deserialisasi nilai pesan Kafka.	String	Tidak	–	Format yang didukung: csv json avro debezium-json canal-json maxwell-json avro-confluent raw Catatan Untuk informasi lebih lanjut, lihat Opsi format.
  key.format	Format untuk serialisasi dan deserialisasi kunci pesan Kafka.	String	Tidak	–	Format yang didukung: csv json avro debezium-json canal-json maxwell-json avro-confluent raw Catatan Saat menggunakan konfigurasi ini, key.options wajib diisi.
  key.fields	Bidang dari skema tabel yang akan digunakan sebagai kunci pesan Kafka.	String	Tidak	–	Pisahkan beberapa nama bidang dengan titik koma (;). Misalnya, 'field1;field2'.
  key.fields-prefix	Awalan kustom untuk semua bidang kunci guna mencegah konflik nama dengan bidang nilai.	String	Tidak	–	Awalan ini digunakan untuk membedakan antara bidang kunci dan bidang nilai. Awalan ini dihapus sebelum serialisasi kunci atau setelah deserialisasi. Catatan Jika Anda menggunakan opsi ini, value.fields-include harus diatur ke EXCEPT_KEY.
  value.format	Format untuk serialisasi dan deserialisasi nilai pesan Kafka.	String	Tidak	–	Konfigurasi ini setara dengan format. Anda hanya dapat mengatur salah satu dari format atau value.format. Jika keduanya dikonfigurasi, value.format akan menggantikan format.
  value.fields-include	Menentukan apakah bidang kunci disertakan dalam format nilai.	String	Tidak	ALL	Nilai yang valid: ALL: Nilai pesan Kafka mencakup semua kolom tabel. EXCEPT_KEY: Nilai pesan Kafka mencakup semua kolom tabel kecuali yang didefinisikan dalam key.fields.

• Tabel sumber

  Opsi	Deskripsi	Jenis	Wajib	Default	Keterangan
  topic	Topik atau topik-topik yang akan dibaca.	String	Tidak	–	Untuk berlangganan ke beberapa topik, pisahkan nama-namanya dengan titik koma (;), misalnya, 'topic-1;topic-2'. Catatan Anda dapat menentukan opsi ini atau topic-pattern, tetapi tidak keduanya.
  topic-pattern	Ekspresi reguler yang cocok dengan topik yang akan di-subscribe. Konsumen berlangganan ke semua topik yang namanya cocok dengan pola ini.	String	Tidak	–	Contoh: user_event_.*: Cocok dengan semua topik yang diawali dengan user_event_. prod\.logs\..*: Cocok dengan topik yang diawali dengan prod.logs. (karakter . harus di-escape). Catatan Anda dapat menentukan opsi ini atau topic, tetapi tidak keduanya.
  properties.group.id	ID kelompok konsumen source Kafka.	String	Tidak	KafkaSource-{Nama-Tabel-Source}	Jika Anda menggunakan ID kelompok konsumen untuk pertama kalinya, Anda juga harus mengatur properties.auto.offset.reset ke earliest atau latest untuk menentukan offset startup awal.
  scan.startup.mode	Offset startup konsumen Kafka.	String	Tidak	group-offsets	Nilai yang valid: earliest-offset: Mulai membaca dari offset paling awal yang tersedia. latest-offset: Mulai membaca dari offset terbaru. group-offsets: Mulai membaca dari offset yang telah dikomit oleh properties.group.id yang ditentukan. timestamp: Mulai membaca dari scan.startup.timestamp-millis yang ditentukan. specific-offsets: Mulai membaca dari offset yang ditentukan dalam scan.startup.specific-offsets. Catatan Opsi ini hanya berlaku saat pekerjaan dimulai tanpa state. Jika pekerjaan dilanjutkan dari checkpoint, pekerjaan tersebut membaca dari offset yang disimpan dalam state checkpoint.
  scan.startup.specific-offsets	Offset awal per-partisi saat scan.startup.mode diatur ke specific-offsets.	String	Tidak	–	Misalnya, partition:0,offset:42;partition:1,offset:300
  scan.startup.timestamp-millis	Timestamp awal dalam milidetik saat scan.startup.mode diatur ke timestamp.	Long	Tidak	–	Unitnya adalah milidetik.
  scan.topic-partition-discovery.interval	Interval penemuan partisi.	Duration	Tidak	5 menit	Connector secara berkala menemukan dan membaca dari partisi baru. Saat Anda menggunakan topic-pattern, connector juga menemukan topik baru yang cocok dengan pola tersebut. Atur interval ke nilai non-positif untuk menonaktifkan fitur ini. Catatan Di Ververica Runtime (VVR) 6.0.x, penemuan partisi dinamis dinonaktifkan secara default. Mulai VVR 8.0, fitur ini diaktifkan secara default dengan interval penemuan 5 menit.
  scan.header-filter	Memfilter pesan berdasarkan header pesan Kafka.	String	Tidak	–	Kunci header dan nilainya dipisahkan oleh titik dua (:). Beberapa kondisi header dihubungkan menggunakan operator logika (& dan |). Operator logika NOT (!) juga didukung. Misalnya, depart:toy|depart:book&!env:test menyimpan data Kafka jika header berisi depart=toy atau depart=book dan tidak berisi env=test. Catatan Opsi ini hanya didukung di Ververica Runtime (VVR) 8.0.6 ke atas. Tanda kurung dalam ekspresi tidak didukung. Operasi logika dievaluasi dari kiri ke kanan. Nilai header dikonversi ke string UTF-8 untuk perbandingan.
  scan.check.duplicated.group.id	Memeriksa apakah sudah ada konsumen aktif lain yang menggunakan properties.group.id.	Boolean	Tidak	false	Nilai yang valid: true: Sebelum memulai pekerjaan, sistem memeriksa duplikasi kelompok konsumen. Jika ditemukan, pekerjaan gagal untuk mencegah konflik. false: Memulai pekerjaan tanpa memeriksa konflik. Catatan Opsi ini hanya didukung di Ververica Runtime (VVR) 6.0.4 ke atas.

• Sink Table

  Opsi	Deskripsi	Jenis	Wajib	Default	Keterangan
  topic	Topik tujuan.	String	Ya	–	–
  sink.partitioner	Memetakan record dari instans sink paralel ke partisi Kafka.	String	Tidak	default	Nilai yang valid: default: Menggunakan partitioner Kafka default. fixed: Setiap instans sink paralel menulis ke partisi Kafka tetap. round-robin: Record didistribusikan ke partisi secara round-robin. Partitioner kustom: Untuk menggunakan partitioner kustom, berikan nama kelas lengkap dari subclass FlinkKafkaPartitioner, misalnya, org.mycompany.MyPartitioner.
  sink.delivery-guarantee	Jaminan pengiriman sink.	String	Tidak	at-least-once	Nilai yang valid: none: Tidak memberikan jaminan apa pun. Record mungkin hilang atau diduplikasi. at-least-once: Menjamin tidak ada record yang hilang, tetapi mungkin terduplikasi. exactly-once: Menggunakan transaksi Kafka untuk memberikan semantik tepat-sekali, memastikan record tidak hilang maupun terduplikasi. Catatan Saat menggunakan semantik exactly-once, Anda juga harus menentukan sink.transactional-id-prefix.
  sink.transactional-id-prefix	Awalan ID transaksi. Diperlukan saat sink.delivery-guarantee diatur ke exactly-once.	String	Ya, jika sink.delivery-guarantee diatur ke exactly-once	–	Hanya diperlukan saat sink.delivery-guarantee diatur ke exactly-once.
  sink.parallelism	Paralelisme operator sink.	Integer	Tidak	–	Secara default, framework menentukan paralelisme berdasarkan operator hulu.

Keamanan dan otentikasi

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',
  /* Konfigurasi SSL */
  /* Path ke truststore untuk sertifikat CA server. */
  /* File yang diunggah menggunakan Artifacts disimpan di direktori /flink/usrlib/. */
  'properties.ssl.truststore.location' = '/flink/usrlib/kafka.client.truststore.jks',
  'properties.ssl.truststore.password' = 'test1234',
  /* Jika otentikasi klien diperlukan, Anda juga harus mengonfigurasi path ke keystore (kunci privat). */
  'properties.ssl.keystore.location' = '/flink/usrlib/kafka.client.keystore.jks',
  'properties.ssl.keystore.password' = 'test1234',
  /* Algoritma yang digunakan untuk memverifikasi hostname server. String kosong menonaktifkan verifikasi hostname. */
  'properties.ssl.endpoint.identification.algorithm' = '',
  /* Konfigurasi SASL */
  /* Atur mekanisme SASL ke SCRAM-SHA-256. */
  'properties.sasl.mechanism' = 'SCRAM-SHA-256',
  /* Konfigurasi JAAS. */
  'properties.sasl.jaas.config' = 'org.apache.flink.kafka.shaded.org.apache.kafka.common.security.scram.ScramLoginModule required username="username" password="password";'
)

Anda dapat menggunakan fitur Artifacts di konsol Realtime Compute for Apache Flink untuk mengunggah sertifikat CA dan kunci privat yang disebutkan dalam contoh. File yang diunggah disimpan di direktori /flink/usrlib. Untuk menggunakan file sertifikat CA bernama my-truststore.jks, Anda dapat mengatur properti 'properties.ssl.truststore.location' di klausa WITH dengan salah satu dari dua cara berikut:

• Atur 'properties.ssl.truststore.location' = '/flink/usrlib/my-truststore.jks'. Metode ini menghindari pengunduhan file secara dinamis dari Object Storage Service (OSS) saat runtime, tetapi tidak mendukung Debug Mode.

• Jika versi mesin Realtime Compute adalah VVR 11.5 atau lebih baru, Anda dapat mengonfigurasi properties.ssl.truststore.location dan properties.ssl.keystore.location ke path OSS absolut. Format path file adalah oss://flink-fullymanaged-<Workspace ID>/artifacts/namespaces/<Nama Namespace>/<nama file>. Metode ini mengunduh file OSS secara dinamis selama runtime Flink dan mendukung Debug Mode.

Offset awal tabel source

CREATE TEMPORARY TABLE kafka_source (
  ...
) WITH (
  'connector' = 'kafka',
  ...
  -- Konsumsi dari offset paling awal.
  'scan.startup.mode' = 'earliest-offset',
  -- Konsumsi dari offset terbaru.
  'scan.startup.mode' = 'latest-offset',
  -- Konsumsi dari offset yang dikomit oleh kelompok konsumen "my-group".
  'properties.group.id' = 'my-group',
  'scan.startup.mode' = 'group-offsets',
  'properties.auto.offset.reset' = 'earliest', -- Jika "my-group" digunakan untuk pertama kali, konsumsi dimulai dari offset paling awal.
  'properties.auto.offset.reset' = 'latest', -- Jika "my-group" digunakan untuk pertama kali, konsumsi dimulai dari offset terbaru.
  -- Konsumsi dari timestamp tertentu dalam milidetik: 1655395200000.
  'scan.startup.mode' = 'timestamp',
  'scan.startup.timestamp-millis' = '1655395200000',
  -- Konsumsi dari offset tertentu.
  'scan.startup.mode' = 'specific-offsets',
  'scan.startup.specific-offsets' = 'partition:0,offset:42;partition:1,offset:300'
);

Meng-commit offset source

Tabel source Kafka hanya meng-commit offset konsumennya ke kluster Kafka setelah checkpoint berhasil, sehingga interval checkpoint yang panjang menyebabkan offset yang dikomit tertinggal. Tabel source menyimpan progres baca aktualnya dalam state checkpoint, yang digunakan sistem untuk pemulihan kesalahan. Offset yang dikomit hanya berfungsi sebagai monitor progres dan tidak digunakan untuk pemulihan, sehingga kegagalan commit tidak memengaruhi akurasi data.

Partitioner sink kustom

Jika strategi partisi bawaan Kafka tidak memenuhi kebutuhan Anda, Anda dapat mengimplementasikan partitioner kustom dengan memperluas kelas FlinkKafkaPartitioner. Setelah pengembangan selesai, kompilasi kode Anda menjadi paket JAR dan unggah menggunakan fitur Artifacts di konsol Realtime Compute. Setelah paket JAR diunggah dan direferensikan, atur parameter sink.partitioner di klausa WITH ke nama kelas lengkap partitioner Anda, misalnya, org.mycompany.MyPartitioner.

Kafka, Upsert Kafka, dan katalog Kafka JSON

Kafka adalah platform streaming event append-only yang tidak mendukung pembaruan atau penghapusan data. Dalam SQL streaming, tabel sink Kafka standar tidak dapat menangani data Change Data Capture (CDC) hulu atau logika retraction dari operator seperti aggregate dan join. Jika Anda perlu menulis data yang berisi perubahan atau retraction, gunakan tabel sink Upsert Kafka.

Untuk menyederhanakan sinkronisasi batch data Change Data Capture (CDC) dari satu atau beberapa tabel database hulu ke Kafka, Anda dapat menggunakan katalog Kafka JSON. Jika data yang disimpan di Kafka dalam format JSON, katalog Kafka JSON memungkinkan Anda melewati langkah mendefinisikan skema dan parameter WITH. Untuk detailnya, lihat Kelola katalog Kafka JSON.

Contoh

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'
    -- Opsional. Meratakan semua kolom bersarang.
);

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');
--Gunakan COALESCE untuk menangani nilai 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'
);

API DataStream

• Buat source Kafka

  Kafka Source menyediakan kelas pembangun untuk membuat Instans Kafka Source. Kode contoh berikut membangun Kafka Source yang mengonsumsi data dari Offset paling awal Topik input-topic. Consumer Group-nya adalah my-group, dan Value Message Kafka dideserialisasi sebagai string.

  Java

  KafkaSource<String> source = KafkaSource.<String>builder()
      .setBootstrapServers(brokers)
      .setTopics("input-topic")
      .setGroupId("my-group")
      .setStartingOffsets(OffsetsInitializer.earliest())
      .setValueOnlyDeserializer(new SimpleStringSchema())
      .build();
  
  env.fromSource(source, WatermarkStrategy.noWatermarks(), "Kafka Source");

  Untuk membangun Kafka Source, Anda harus menentukan properti berikut.

  Parameter	Deskripsi
  BootstrapServers	Daftar alamat broker Kafka. Atur properti ini dengan memanggil metode setBootstrapServers(String).
  GroupId	ID Kelompok Konsumen. Atur properti ini dengan memanggil metode setGroupId(String).
  Topik atau Partisi	Topik atau partisi untuk berlangganan. Kafka Source mendukung tiga metode berikut untuk berlangganan ke topik atau partisi: Berlangganan ke semua partisi dari topik dalam daftar. KafkaSource.builder().setTopics("topic-a","topic-b") Pola topik: Berlangganan ke semua partisi dari topik yang namanya cocok dengan ekspresi reguler yang ditentukan. KafkaSource.builder().setTopicPattern("topic.*") Daftar partisi, di mana Anda dapat berlangganan ke partisi tertentu. final HashSet<TopicPartition> partitionSet = new HashSet<>(Arrays.asList( new TopicPartition("topic-a", 0), // Partisi 0 dari topik "topic-a" new TopicPartition("topic-b", 5))); // Partisi 5 dari topik "topic-b" KafkaSource.builder().setPartitions(partitionSet)
  Deserializer	Deserializer yang digunakan untuk mengurai pesan Kafka. Tentukan deserializer menggunakan metode setDeserializer(KafkaRecordDeserializationSchema). KafkaRecordDeserializationSchema mendefinisikan cara mengurai ConsumerRecord Kafka. Jika Anda hanya perlu mengurai Nilai Pesan Kafka, Anda dapat menggunakan salah satu metode berikut: Gunakan metode setValueOnlyDeserializer(DeserializationSchema) dari kelas builder. DeserializationSchema mendefinisikan cara mengurai data biner Nilai Pesan Kafka. Gunakan kelas yang mengimplementasikan antarmuka Deserializer Kafka. Misalnya, Anda dapat menggunakan StringDeserializer untuk mengurai Nilai Pesan Kafka menjadi string. import org.apache.kafka.common.serialization.StringDeserializer; KafkaSource.<String>builder() .setDeserializer(KafkaRecordDeserializationSchema.valueOnly(StringDeserializer.class)); Catatan Untuk mengurai ConsumerRecord lengkap, Anda harus mengimplementasikan antarmuka KafkaRecordDeserializationSchema.

  POM

  Kafka DataStream Connector tersedia di repositori pusat Maven.

  <dependency>
      <groupId>com.alibaba.ververica</groupId>
      <artifactId>ververica-connector-kafka</artifactId>
      <version>${vvr-version}</version>
  </dependency>

  Saat menggunakan DataStream Connector Kafka, pertimbangkan properti berikut:

  • Offset awal

    Kafka Source menentukan offset awal-nya menggunakan initializer offset (OffsetsInitializer). Initializer bawaan meliputi:

    Offset initializer	Kode
    Memulai konsumsi dari Offset paling awal.	KafkaSource.builder().setStartingOffsets(OffsetsInitializer.earliest())
    Memulai konsumsi dari Offset terbaru.	KafkaSource.builder().setStartingOffsets(OffsetsInitializer.latest())
    Memulai konsumsi data yang timestamp-nya lebih besar dari atau sama dengan waktu yang ditentukan. Unitnya adalah milidetik.	KafkaSource.builder().setStartingOffsets(OffsetsInitializer.timestamp(1592323200000L))
    Memulai konsumsi dari Offset yang dikomit oleh Kelompok Konsumen. Jika tidak ada Offset yang dikomit, menggunakan strategi reset yang ditentukan (misalnya, Offset paling awal).	KafkaSource.builder().setStartingOffsets(OffsetsInitializer.committedOffsets(OffsetResetStrategy.EARLIEST))
    Konsumsi dimulai dari offset yang dikomit oleh kelompok konsumen, dan tidak ada kebijakan reset offset yang ditentukan.	KafkaSource.builder().setStartingOffsets(OffsetsInitializer.committedOffsets())

    Catatan

    • Jika initializer bawaan tidak memenuhi kebutuhan Anda, Anda dapat mengimplementasikan initializer offset kustom.

    • Jika Anda tidak menentukan initializer offset, nilai default adalah OffsetsInitializer.earliest().

  • Mode streaming dan mode batch

    Kafka Source mendukung mode streaming dan Batch Mode. Secara default, beroperasi dalam mode streaming, di mana Job berjalan tanpa batas hingga gagal atau dibatalkan. Untuk mengonfigurasi Kafka Source agar berjalan dalam Batch Mode, Anda dapat menggunakan setBounded(OffsetsInitializer) untuk menentukan Offset berhenti. Kafka Source keluar ketika semua partisi mencapai offset berhenti yang ditentukan.

    Catatan

    Kafka Source dalam mode streaming biasanya tidak memiliki Offset berhenti. Namun, untuk tujuan pengujian, Anda dapat menggunakan setUnbounded(OffsetsInitializer) untuk menentukan Offset berhenti bahkan dalam mode streaming. Perhatikan perbedaan nama metode untuk menentukan Offset berhenti: setUnbounded untuk mode streaming dan setBounded untuk Batch Mode.

  • Penemuan partisi dinamis

    Untuk menangani penskalaan Topik atau pembuatan topik baru tanpa me-restart Job Flink, Anda dapat mengaktifkan Penemuan partisi dinamis saat berlangganan ke topik berdasarkan pola. Fitur ini dinonaktifkan secara default dan harus diaktifkan secara eksplisit:

    KafkaSource.builder()
        .setProperty("partition.discovery.interval.ms", "10000") // Temukan partisi baru setiap 10 detik.

    Penting

    Fitur penemuan partisi dinamis bergantung pada mekanisme pembaruan metadata kluster Kafka. Jika kluster Kafka tidak memperbarui informasi partisi secara tepat waktu, partisi baru mungkin tidak ditemukan. Pastikan konfigurasi partition.discovery.interval.ms kluster Kafka sesuai dengan skenario aktual Anda.

  • Waktu event dan watermark

    Secara default, Kafka Source menggunakan timestamp dari Pesan Kafka sebagai Event Time. Anda dapat mendefinisikan strategi Watermark kustom untuk mengekstrak Event Time dari isi Pesan dan mengirimkan Watermark ke hulu.

    env.fromSource(kafkaSource, new CustomWatermarkStrategy(), "Kafka Source With Custom Watermark Strategy")

    Untuk mempelajari lebih lanjut tentang strategi Watermark kustom, lihat Generating Watermarks.

    Catatan

    Jika subtugas sumber menganggur—misalnya, ketika partisi Kafka tidak memiliki data baru atau paralelisme sumber lebih tinggi daripada jumlah partisi Kafka—Watermark untuk subtugas tersebut tidak akan maju, yang dapat memblokir komputasi jendela hilir.

    Untuk mengatasi masalah ini, pertimbangkan solusi berikut:

    • Konfigurasikan timeout idle sumber: Aktifkan properti table.exec.source.idle-timeout untuk menandai sumber yang idle sebagai sementara idle, sehingga memungkinkan Watermark hilir maju.

    • Atur Paralelisme yang sesuai: Pastikan Paralelisme source tidak lebih besar daripada jumlah partisi Kafka.

  • Offset commit

    Saat checkpoint diaktifkan, Kafka Source meng-commit Offset konsumen saat ini ke Kafka ketika Checkpoint selesai. Hal ini memastikan state Checkpoint Flink konsisten dengan Offset yang dikomit di broker Kafka. Jika checkpoint dinonaktifkan, Kafka Source mengandalkan mekanisme commit periodik otomatis internal konsumen Kafka. Fitur ini dikontrol oleh properti konsumen Kafka enable.auto.commit dan auto.commit.interval.ms.

    Catatan

    Kafka Source tidak mengandalkan offset yang dikomit untuk toleransi kesalahan dan pemulihan. Meng-commit offset hanya untuk memantau progres konsumen Kafka dan Kelompok Konsumen.

  • Properti lainnya

    Selain properti yang disebutkan, Anda dapat menggunakan setProperties(Properties) dan setProperty(String, String) untuk mengatur properti apa pun untuk Kafka Source dan konsumen Kafka yang mendasarinya. Kafka Source menyediakan properti spesifik berikut.

    Parameter	Deskripsi
    client.id.prefix	Awalan ID klien untuk konsumen Kafka.
    partition.discovery.interval.ms	Interval penemuan partisi dalam milidetik. Nilai -1 menonaktifkan penemuan partisi dinamis. Catatan Di Batch Mode, properti ini secara otomatis diatur ke -1.
    register.consumer.metrics	Mendaftarkan metrik konsumen Kafka di Flink.
    Konfigurasi Konsumen Kafka Lainnya	Untuk daftar lengkap konfigurasi konsumen Kafka, lihat dokumentasi resmi Apache Kafka. Penting Untuk memastikan operasi yang benar, DataStream Connector Kafka menimpa properti yang dikonfigurasi secara manual berikut: key.deserializer selalu ditimpa ke org.apache.kafka.common.serialization.ByteArrayDeserializer. value.deserializer selalu ditimpa ke org.apache.kafka.common.serialization.ByteArrayDeserializer. auto.offset.reset.strategy ditimpa oleh strategi yang disediakan oleh OffsetsInitializer.

    Contoh berikut menunjukkan cara mengonfigurasi konsumen Kafka untuk menggunakan mekanisme SASL PLAIN dan menyediakan konfigurasi JAAS.

    KafkaSource.builder()
        .setProperty("sasl.mechanism", "PLAIN")
        .setProperty("sasl.jaas.config", "org.apache.kafka.common.security.plain.PlainLoginModule required username=\"username\" password=\"password\";")

  • Pemantauan

    Kafka Source mengekspos metrik melalui sistem metrik Flink untuk pemantauan dan diagnostik.

    • Cakupan metrik

      Semua metrik untuk pembaca source Kafka didaftarkan di bawah grup metrik KafkaSourceReader, yang merupakan subgrup dari grup metrik operator. Metrik yang terkait dengan partisi topik tertentu didaftarkan di subgrup KafkaSourceReader.topic.<topic_name>.partition.<partition_id>.

      Misalnya, metrik Current Consumer Offset (currentOffset) untuk Partisi 1 dari topik "my-topic" tersedia di .operator.KafkaSourceReader.topic.my-topic.partition.1.currentOffset. Jumlah commit yang berhasil (commitsSucceeded) tersedia di .operator.KafkaSourceReader.commitsSucceeded.

    • Daftar metrik

      Metrik	Deskripsi	Cakupan
      currentOffset	Offset konsumen saat ini dari partisi.	TopicPartition
      committedOffset	Offset terakhir yang dikomit untuk partisi.	TopicPartition
      commitsSucceeded	Jumlah total commit offset yang berhasil.	KafkaSourceReader
      commitsFailed	Jumlah commit yang gagal	KafkaSourceReader

    • Metrik konsumen Kafka

      Metrik konsumen Kafka yang mendasari didaftarkan di grup metrik KafkaSourceReader.KafkaConsumer. Misalnya, metrik records-consumed-total didaftarkan di .operator.KafkaSourceReader.KafkaConsumer.records-consumed-total.

      Anda dapat menggunakan properti register.consumer.metrics untuk menentukan apakah akan mendaftarkan metrik konsumen Kafka. Opsi ini diaktifkan secara default (true). Untuk informasi lebih lanjut tentang metrik konsumen Kafka, lihat dokumentasi Apache Kafka.

• Buat sink Kafka

  Kafka Sink Flink menulis aliran data ke satu atau beberapa topik Kafka.

  DataStream<String> stream = ...
  
  Properties kafkaProperties = new Properties();
  kafkaProperties.setProperty("bootstrap.servers", "localhost:9092");
  
  KafkaSink<String> sink = KafkaSink.<String>builder()
          .setKafkaProducerConfig(kafkaProperties)
          .setRecordSerializer(
                  KafkaRecordSerializationSchema.builder()
                          .setTopic("my-topic")
                          .setValueSerializationSchema(new SimpleStringSchema())
                          .build())
          .setDeliveryGuarantee(DeliveryGuarantee.AT_LEAST_ONCE)
          .build();
  
  stream.sinkTo(sink);

  Untuk membangun Kafka Sink, Anda harus mengonfigurasi properti berikut.

  Parameter	Deskripsi
  Properti klien Kafka	Properti bootstrap.servers wajib diisi. Properti ini menentukan daftar broker Kafka yang dipisahkan koma.
  Serializer record	Anda harus menyediakan KafkaRecordSerializationSchema untuk mengonversi data input menjadi ProducerRecord Kafka. Flink menyediakan builder skema yang menawarkan komponen umum, seperti serialisasi untuk kunci dan nilai pesan, pemilihan topik, dan partisi pesan. Anda juga dapat mengimplementasikan antarmuka yang sesuai untuk kontrol lebih rinci. Metode ProducerRecord<byte[], byte[]> serialize(T element, KafkaSinkContext context, Long timestamp) dipanggil untuk setiap record masuk untuk menghasilkan ProducerRecord yang akan ditulis ke Kafka. ProducerRecord memberikan kontrol rinci tentang cara setiap record ditulis ke Kafka, memungkinkan Anda untuk: Tetapkan Topic tujuan. Atur Message Key. Menentukan Partisi tujuan.
  Jaminan pengiriman	Parameter bootstrap.servers wajib diisi dan menentukan daftar broker Kafka yang dipisahkan koma.
  Jaminan pengiriman	Saat checkpoint Flink diaktifkan, Flink Kafka Sink dapat memberikan semantik tepat-sekali. Selain mengaktifkan checkpoint, Anda dapat menggunakan parameter DeliveryGuarantee untuk menentukan jaminan pengiriman yang berbeda. Parameter DeliveryGuarantee menyediakan opsi berikut: DeliveryGuarantee.NONE: (Default) Flink tidak memberikan jaminan apa pun. Data mungkin hilang atau diduplikasi. DeliveryGuarantee.AT_LEAST_ONCE: Menjamin tidak ada data yang hilang, tetapi duplikasi mungkin terjadi. DeliveryGuarantee.EXACTLY_ONCE: Menggunakan transaksi Kafka untuk memberikan semantik tepat-sekali. Catatan Saat menggunakan semantik EXACTLY_ONCE, lihat Pertimbangan untuk semantik tepat-sekali.

Toleransi dan pengumpulan data kotor

Sumber data Kafka Anda mungkin berisi record yang rusak, yang biasa disebut data kotor. Untuk mencegah pekerjaan Anda gagal dan restart berulang kali, Anda dapat mengonfigurasinya untuk melewatkan record yang tidak valid ini. Misalnya:

source:
  type: kafka
  name: Kafka Source
  properties.bootstrap.servers: host:9092
  topic: test-topic
  value.format: json
  scan.startup.mode: earliest-offset
  # Aktifkan Toleransi Data Kotor
  ingestion.ignore-errors: true
  # Toleransi hingga 1000 record data kotor
  ingestion.error-tolerance.max-count: 1000

Dengan konfigurasi ini, pekerjaan terus berjalan selama menemui tidak lebih dari 1.000 record kotor. Setelah jumlah tersebut melebihi ambang batas, pekerjaan gagal sehingga Anda dapat menyelidiki data Anda.

Untuk memastikan pekerjaan Anda tidak pernah gagal karena data kotor, gunakan konfigurasi berikut:

source:
  type: kafka
  name: Kafka Source
  properties.bootstrap.servers: host:9092
  topic: test-topic
  value.format: json
  scan.startup.mode: earliest-offset
  # Aktifkan Toleransi Data Kotor
  ingestion.ignore-errors: true
  # Toleransi semua record data kotor
  ingestion.error-tolerance.max-count: -1

Meskipun toleransi data kotor menjaga pekerjaan Anda tetap berjalan, Anda mungkin juga ingin memeriksa record yang bermasalah. Anda mungkin juga ingin menganalisis data kotor untuk meningkatkan produsen Kafka Anda. Seperti dijelaskan dalam Pengumpulan Data Kotor, Anda dapat melihat data kotor pekerjaan di log TaskManager. Misalnya:

source:
  type: kafka
  name: Kafka Source
  properties.bootstrap.servers: host:9092
  topic: test-topic
  value.format: json
  scan.startup.mode: earliest-offset
  # Aktifkan Toleransi Data Kotor
  ingestion.ignore-errors: true
  # Toleransi semua record data kotor
  ingestion.error-tolerance.max-count: -1

pipeline:
  dirty-data.collector:
    # Tulis data kotor ke file log TaskManager
    type: logger

Pemetaan nama tabel dan topik

Saat Kafka berfungsi sebagai sink Flink CDC, format pesan (seperti Debezium JSON atau Canal JSON) menyematkan nama tabel asli. Konsumen hilir umumnya menggunakan nama yang disematkan tersebut sebagai pengenal tabel alih-alih nama topik, sehingga penting untuk mengonfigurasi pemetaan antara nama tabel dan topik secara tepat.

Anggaplah Anda perlu menyinkronkan dua tabel dari database MySQL: mydb.mytable1 dan mydb.mytable2. Strategi pemetaan berikut tersedia:

1. Tanpa strategi pemetaan

Tanpa strategi pemetaan apa pun, data untuk setiap tabel ditulis ke topik yang dinamai dalam format <Nama Database>.<Nama Tabel>. Oleh karena itu, data dari mydb.mytable1 ditulis ke topik bernama mydb.mytable1, dan data dari mydb.mytable2 ditulis ke topik bernama mydb.mytable2. Berikut adalah contoh konfigurasi:

source:
  type: mysql
  name: MySQL Source
  hostname: ${secret_values.mysql.hostname}
  port: ${mysql.port}
  username: ${secret_values.mysql.username}
  password: ${secret_values.mysql.password}
  tables: mydb.mytable1,mydb.mytable2
  server-id: 8601-8604

sink:
  type: kafka
  name: Kafka Sink
  properties.bootstrap.servers: ${kafka.bootstraps.server}

2. Pemetaan aturan rute (Tidak disarankan)

Anda mungkin ingin menulis data ke topik tertentu daripada menggunakan format default <Nama Database>.<Nama Tabel>. Untuk melakukan ini, Anda dapat mengonfigurasi aturan rute. Berikut adalah contoh konfigurasi:

source:
  type: mysql
  name: MySQL Source
  hostname: ${secret_values.mysql.hostname}
  port: ${mysql.port}
  username: ${secret_values.mysql.username}
  password: ${secret_values.mysql.password}
  tables: mydb.mytable1,mydb.mytable2
  server-id: 8601-8604

sink:
  type: kafka
  name: Kafka Sink
  properties.bootstrap.servers: ${kafka.bootstraps.server}
  
 route:
  - source-table: mydb.mytable1,mydb.mytable2
    sink-table: mytable

Dalam hal ini, semua data dari mydb.mytable1 dan mydb.mytable2 ditulis ke satu topik bernama mytable.

Namun, aturan rute yang mengubah topik tujuan juga mengubah nama tabel dalam pesan Kafka (dalam format Debezium JSON atau Canal JSON). Nama tabel dalam semua pesan Kafka menjadi mytable. Hal ini dapat menyebabkan perilaku tak terduga pada sistem yang mengonsumsi pesan dari topik ini.

3. Pemetaan dengan sink.tableId-to-topic.mapping (Direkomendasikan)

Untuk memetakan nama tabel ke topik sambil mempertahankan nama tabel sumber asli, gunakan parameter sink.tableId-to-topic.mapping. Berikut adalah contoh konfigurasi:

source:
  type: mysql
  name: MySQL Source
  hostname: ${secret_values.mysql.hostname}
  port: ${mysql.port}
  username: ${secret_values.mysql.username}
  password: ${secret_values.mysql.password}
  tables: mydb.mytable1,mydb.mytable2
  server-id: 8601-8604
  sink.tableId-to-topic.mapping: mydb.mytable1,mydb.mytable2:mytable

sink:
  type: kafka
  name: Kafka Sink
  properties.bootstrap.servers: ${kafka.bootstraps.server}

Atau, Anda dapat menggunakan konfigurasi berikut:

source:
  type: mysql
  name: MySQL Source
  hostname: ${secret_values.mysql.hostname}
  port: ${mysql.port}
  username: ${secret_values.mysql.username}
  password: ${secret_values.mysql.password}
  tables: mydb.mytable1,mydb.mytable2
  server-id: 8601-8604
  sink.tableId-to-topic.mapping: mydb.mytable1:mytable;mydb.mytable2:mytable

sink:
  type: kafka
  name: Kafka Sink
  properties.bootstrap.servers: ${kafka.bootstraps.server}

Dalam kasus ini, seluruh data dari mydb.mytable1 dan mydb.mytable2 ditulis ke topik mytable, sedangkan nama tabel dalam pesan Kafka (dalam format JSON Debezium atau JSON Canal) tetap dipertahankan sebagai mydb.mytable1 atau mydb.mytable2. Dengan demikian, sistem downstream tetap dapat mengidentifikasi tabel sumber asli dari setiap catatan.

Semantik tepat-sekali

• Konfigurasikan tingkat isolasi konsumen

  Semua aplikasi yang mengonsumsi data Kafka harus mengatur properti isolation.level:

  • read_committed: Hanya membaca data yang telah dikomit.

  • read_uncommitted (Default): Dapat membaca data yang belum dikomit.

  EXACTLY_ONCE bergantung pada read_committed. Jika tidak, konsumen mungkin melihat data yang belum dikomit, sehingga merusak konsistensi.

• Timeout transaksi dan kehilangan data

  Saat memulihkan dari checkpoint, Realtime Compute for Apache Flink hanya mempertimbangkan transaksi yang telah dikomit sebelum checkpoint tersebut dimulai. Jika durasi antara kegagalan pekerjaan dan restart-nya melebihi timeout transaksi Kafka, Kafka secara otomatis membatalkan transaksi yang terbuka, yang dapat mengakibatkan kehilangan data.

  • transaction.max.timeout.ms default untuk broker Kafka adalah 15 menit.

  • Secara default, Flink Kafka Sink mengatur parameter transaction.timeout.ms ke 1 jam.

  • Anda harus menaikkan transaction.max.timeout.ms pada broker agar lebih besar dari atau sama dengan pengaturan di Flink.

• Kolam produsen dan checkpoint konkuren

  Mode EXACTLY_ONCE menggunakan kolam produsen Kafka berukuran tetap. Setiap checkpoint menggunakan satu produsen dari kolam ini. Jika jumlah checkpoint konkuren melebihi ukuran kolam, pekerjaan gagal.

  Konfigurasikan ukuran kolam produsen berdasarkan jumlah maksimum checkpoint konkuren.

• Batasan penskalaan turun paralelisme

  Jika pekerjaan gagal sebelum checkpoint pertama selesai, informasi kolam produsen asli hilang saat restart. Oleh karena itu, jangan menurunkan paralelisme pekerjaan sebelum checkpoint pertama selesai. Jika penskalaan turun diperlukan, paralelisme baru tidak boleh kurang dari FlinkKafkaProducer.SAFE_SCALE_DOWN_FACTOR.

• Transaksi memblokir pembacaan

  Dalam mode read_committed, transaksi apa pun yang belum dikomit atau dibatalkan akan memblokir operasi baca pada seluruh topik.

  Misalnya:

  • Transaksi 1 menulis data.

  • Transaksi 2 menulis lebih banyak data dan dikomit.

  • Selama Transaksi 1 tetap terbuka, data dari Transaksi 2 yang telah dikomit tidak terlihat oleh konsumen.

  Hal ini memiliki implikasi berikut:

  • Selama operasi normal, latensi visibilitas data kira-kira sama dengan interval checkpoint.

  • Jika pekerjaan gagal, topik apa pun yang sedang ditulisnya akan diblokir untuk konsumen hingga pekerjaan tersebut dijalankan ulang atau transaksi timeout. Dalam kasus ekstrem, proses timeout transaksi itu sendiri juga dapat memengaruhi operasi baca.

FAQ

• Bagaimana cara mengambil data JSON dari Kafka menggunakan Realtime Compute for Apache Flink?

• Mengapa Window berbasis Event Time tidak menghasilkan output untuk Tabel Source ApsaraMQ for Kafka?

• Bagaimana cara menghubungkan ke Kluster Kafka yang diamankan?

• Bagaimana cara menyelesaikan konflik nama bidang?

• Ruang Kerja dan Namespace