Realtime Compute for Apache Flink:Message Queue for Kafka

Latar Belakang

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', konektor Kafka mengaktifkan penulisan transaksional, dengan masalah yang diketahui sebagai berikut:

• ID Transaksi baru dihasilkan untuk setiap checkpoint. Jika interval checkpoint terlalu pendek, jumlah ID Transaksi yang berlebihan dapat dibuat. Hal ini dapat menyebabkan koordinator di kluster Apache Kafka kehabisan memori, sehingga mengganggu stabilitas kluster.

• Instance Producer dibuat untuk setiap transaksi. Jika terlalu banyak transaksi yang dikomit secara bersamaan, TaskManager dapat kehabisan memori, sehingga mengganggu stabilitas 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 konektor Upsert Kafka untuk menulis ke tabel kunci primer, 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

Konektor Kafka dapat digunakan sebagai tabel sumber 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',
  ...
);

DENGAN parameter

• Umum

  Parameter	Deskripsi	Tipe	Wajib	Default	Keterangan
  connector	Konektor yang digunakan.	String	Ya	N/A	Nilainya harus kafka.
  properties.bootstrap.servers	Daftar alamat broker Kafka.	String	Ya	N/A	Format: host:port,host:port,.... Pisahkan alamat dengan koma (,).
  properties.*	Properti tambahan untuk klien Kafka.	String	Tidak	N/A	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 prefiks 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. Anda tidak dapat menggunakan metode ini untuk mengonfigurasi opsi berikut karena Konektor Kafka akan menimpanya: key.deserializer value.deserializer
  format	Format untuk serialisasi dan deserialisasi nilai pesan Kafka.	String	Tidak	N/A	Format yang didukung: csv json avro debezium-json canal-json maxwell-json avro-confluent raw Catatan Untuk informasi selengkapnya, lihat Format options.
  key.format	Format untuk serialisasi dan deserialisasi kunci pesan Kafka.	String	Tidak	N/A	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 digunakan sebagai kunci pesan Kafka.	String	Tidak	N/A	Pisahkan beberapa nama bidang dengan titik koma (;). Misalnya, 'field1;field2'.
  key.fields-prefix	Prefiks khusus untuk semua bidang kunci guna mencegah konflik nama dengan bidang nilai.	String	Tidak	N/A	Prefiks ini digunakan untuk membedakan antara bidang kunci dan bidang nilai. Prefiks 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	N/A	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

  Parameter	Deskripsi	Type	Wajib	Default	Keterangan
  topic	Topik atau topik yang akan dibaca.	String	Tidak	N/A	Untuk berlangganan beberapa topik, pisahkan nama mereka 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 menentukan topik mana yang akan berlangganan. Konsumen berlangganan ke semua topik yang namanya sesuai dengan pola ini.	String	Tidak	N/A	Catatan Anda dapat menentukan opsi ini atau topic, tetapi tidak keduanya.
  properties.group.id	ID Kelompok Konsumen untuk sumber Kafka.	String	Tidak	KafkaSource-{Nama-Tabel-Sumber}	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 untuk 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 dikomit dari 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 dari State bersih. Jika Pekerjaan dimulai ulang dari Checkpoint, selalu dilanjutkan dari Offset yang disimpan dalam State Checkpoint.
  scan.startup.specific-offsets	Offset awal untuk setiap partisi saat scan.startup.mode adalah specific-offsets.	String	Tidak	N/A	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	N/A	Unitnya adalah milidetik.
  scan.topic-partition-discovery.interval	Interval untuk menemukan partisi baru dalam topik.	Duration	Tidak	5 menit	Konektor secara berkala menemukan dan membaca dari partisi baru. Saat menggunakan topic-pattern, Konektor juga menemukan topik baru yang sesuai 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 dari VVR 8.0, fitur ini diaktifkan secara default dengan interval penemuan 5 menit.
  scan.header-filter	Memfilter pesan berdasarkan header pesan Kafka.	String	Tidak	N/A	Kunci header dan nilainya dipisahkan dengan 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 dan versi setelahnya. 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	Menentukan apakah akan memeriksa apakah Konsumen aktif lain 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 dan versi setelahnya.

• Tabel sink

  Parameter	Deskripsi	Type	Wajib	Default	Keterangan
  topic	Topik yang akan ditulis.	String	Ya	N/A	N/A
  sink.partitioner	Menentukan cara memetakan catatan dari instance sink paralel ke partisi Kafka.	String	Tidak	default	Nilai yang valid: default: Menggunakan partitioner Kafka default. fixed: Setiap instance sink paralel menulis ke partisi Kafka tetap. round-robin: Catatan didistribusikan ke partisi secara round-robin. Partitioner khusus: Untuk menggunakan partitioner khusus, berikan nama kelas lengkap dari subclass FlinkKafkaPartitioner, misalnya, org.mycompany.MyPartitioner.
  sink.delivery-guarantee	Jaminan pengiriman untuk sink.	String	Tidak	at-least-once	Nilai yang valid: none: Tidak memberikan jaminan. Catatan mungkin hilang atau diduplikasi. at-least-once: Menjamin bahwa tidak ada catatan yang hilang, tetapi mungkin diduplikasi. exactly-once: Menggunakan transaksi Kafka untuk memberikan Semantik tepat-sekali, memastikan catatan tidak hilang maupun diduplikasi. Catatan Saat menggunakan Semantik exactly-once, Anda juga harus menentukan sink.transactional-id-prefix.
  sink.transactional-id-prefix	Prefiks untuk ID transaksi, wajib saat sink.delivery-guarantee adalah exactly-once.	String	Ya, jika sink.delivery-guarantee adalah exactly-once	N/A	Wajib hanya saat sink.delivery-guarantee diatur ke exactly-once.
  sink.parallelism	Paralelisme Operator sink.	Integer	Tidak	N/A	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 dari Konsol Komputasi Real-time 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 Mode Debug.

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

Offset awal tabel sumber

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 kelompok konsumen "my-group".
  'properties.group.id' = 'my-group',
  'scan.startup.mode' = 'group-offsets',
  'properties.auto.offset.reset' = 'earliest', -- Jika "my-group" digunakan pertama kali, konsumsi dimulai dari offset paling awal.
  'properties.auto.offset.reset' = 'latest', -- Jika "my-group" digunakan 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'
);

Mengkomit offset sumber

Tabel sumber Kafka hanya mengkomit offset konsumen saat ini ke kluster Kafka setelah checkpoint berhasil. Jika interval checkpoint panjang, offset konsumen di kluster Kafka tertinggal. Selama checkpoint, tabel sumber Kafka menyimpan progres bacaannya saat ini dalam statusnya. Sistem menggunakan status ini untuk pemulihan kesalahan dan tidak bergantung pada offset yang dikomit ke kluster Kafka. Offset dikomit hanya untuk memantau progres bacaan di Kafka. Kegagalan komit offset tidak memengaruhi akurasi data.

Pemartisi sink kustom

Jika strategi partisi bawaan Kafka tidak memenuhi kebutuhan Anda, Anda dapat mengimplementasikan partitioner khusus 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 sistem antrian pesan 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 penarikan kembali operator seperti agregat dan join. Jika Anda perlu menulis data yang berisi perubahan atau penarikan kembali, 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'
);

DataStream API

• Buat Kafka Source

  Kafka Source menyediakan kelas builder untuk membuat instance Kafka Source. Kode contoh berikut membangun Kafka Source yang mengonsumsi data dari Offset paling awal dari topik input-topic. Kelompok Konsumen adalah my-group, dan Nilai Pesan 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 topik dalam daftar. KafkaSource.builder().setTopics("topic-a","topic-b") Pola topik: Berlangganan ke semua partisi topik yang namanya cocok dengan ekspresi reguler yang ditentukan. KafkaSource.builder().setTopicPattern("topic.*") Daftar partisi, tempat 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 deserialisasi menggunakan metode setDeserializer(KafkaRecordDeserializationSchema). KafkaRecordDeserializationSchema mendefinisikan cara mengurai ConsumerRecord Kafka. Jika Anda hanya perlu mengurai Value dari Message Kafka, Anda dapat menggunakan salah satu metode berikut: Gunakan metode setValueOnlyDeserializer(DeserializationSchema) dari kelas pembuat. DeserializationSchema menentukan cara mengurai data biner dari Message Value 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

  Konektor DataStream Kafka 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 awalnya menggunakan Offset Initializer (OffsetsInitializer). Inisialisasi bawaan meliputi:

    Inisialisasi offset	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 inisialisasi bawaan tidak memenuhi kebutuhan Anda, Anda dapat mengimplementasikan Offset Initializer khusus.

    • Jika Anda tidak menentukan Offset Initializer, default-nya adalah OffsetsInitializer.earliest().

  • Mode Streaming dan Mode Batch

    Kafka Source mendukung Mode Streaming dan Mode Batch. Secara default, beroperasi dalam Mode Streaming, di mana Pekerjaan berjalan tanpa henti hingga gagal atau dibatalkan. Untuk mengonfigurasi Kafka Source agar berjalan dalam Mode Batch, Anda dapat menggunakan setBounded(OffsetsInitializer) untuk menentukan Offset berhenti. Kafka Source keluar saat 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 Mode Batch.

  • Penemuan Partisi Dinamis

    Untuk menangani penskalaan Topik atau pembuatan topik baru tanpa memulai ulang Pekerjaan 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 Peristiwa dan Watermark

    Secara default, Kafka Source menggunakan timestamp dari Pesan Kafka sebagai Waktu Event. Anda dapat mendefinisikan strategi Watermark khusus untuk mengekstrak Waktu Event 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 khusus, lihat Generating Watermarks.

    Catatan

    Jika subtugas sumber tidak aktif—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 menghambat komputasi jendela downstream.

    Untuk mengatasi masalah ini, pertimbangkan solusi berikut:

    • Konfigurasikan waktu tunggu idle sumber: Aktifkan properti table.exec.source.idle-timeout untuk menandai sumber idle sebagai sementara idle. Hal ini memungkinkan Watermark hilir maju.

    • Atur Paralelisme yang sesuai: Pastikan Paralelisme sumber tidak lebih besar dari jumlah partisi Kafka.

  • Offset Commit

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

    Catatan

    Kafka Source tidak bergantung pada offset yang dikomit untuk toleransi kesalahan dan pemulihan. Mengkomit 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	Menentukan prefiks ID klien untuk konsumen Kafka.
    partition.discovery.interval.ms	Menentukan interval dalam milidetik untuk menemukan partisi baru. Nilai -1 menonaktifkan penemuan partisi dinamis. Catatan Dalam Batch Mode, properti ini secara otomatis diatur ke -1.
    register.consumer.metrics	Menentukan apakah akan 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 sumber Kafka terdaftar di bawah grup metrik KafkaSourceReader, yang merupakan subgrup dari grup metrik operator. Metrik yang terkait dengan Partisi Topik tertentu terdaftar di subgrup KafkaSourceReader.topic.<topic_name>.partition.<partition_id>.

      Misalnya, metrik Offset Konsumen saat ini (currentOffset) untuk Partisi 1 dari topik "my-topic" tersedia di .operator.KafkaSourceReader.topic.my-topic.partition.1.currentOffset. Jumlah komit yang berhasil (commitsSucceeded) tersedia di .operator.KafkaSourceReader.commitsSucceeded.

    • Daftar metrik

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

    • Metrik konsumen Kafka

      Metrik konsumen Kafka yang mendasari terdaftar dalam kelompok metrik KafkaSourceReader.KafkaConsumer. Sebagai contoh, metrik records-consumed-total Metrik terdaftar di .operator.KafkaSourceReader.KafkaConsumer.records-consumed-total.

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

• Buat Kafka Sink

  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 Properti wajib diisi. Properti ini menentukan daftar broker Kafka yang dipisahkan koma.
  Serializer catatan	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 catatan masuk untuk menghasilkan ProducerRecord yang akan ditulis ke Kafka. ProducerRecord memberikan kontrol rinci tentang cara setiap catatan ditulis ke Kafka, memungkinkan Anda untuk: Atur 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, Kafka Sink Flink 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. Data mungkin hilang atau diduplikasi. DeliveryGuarantee.AT_LEAST_ONCE: Menjamin bahwa 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 EXACTLY_ONCE.

Toleransi dan pengumpulan data kotor

Sumber Data Kafka Anda mungkin berisi catatan yang rusak, juga dikenal sebagai Data Kotor. Untuk mencegah Pekerjaan Anda sering gagal dan dimulai ulang, Anda dapat mengonfigurasinya untuk mengabaikan catatan 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 catatan data kotor
  ingestion.error-tolerance.max-count: 1000

Konfigurasi ini mentoleransi hingga 1.000 catatan kotor, memungkinkan Pekerjaan berjalan normal. Jika jumlah catatan kotor melebihi ambang batas ini, Pekerjaan gagal, mendorong Anda untuk memvalidasi 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 catatan data kotor
  ingestion.error-tolerance.max-count: -1

Kebijakan Toleransi Data Kotor memastikan Pekerjaan Anda tidak gagal karena catatan tidak valid. Anda mungkin juga ingin menganalisis Data Kotor ini untuk menyesuaikan perilaku produsen data 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 catatan 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 menggunakan Kafka sebagai Sink untuk Job Pengambilan Data, format pesan—seperti JSON Debezium atau JSON Canal—mencakup informasi nama tabel asli. Sistem downstream yang mengonsumsi pesan tersebut umumnya menggunakan nama tabel yang disematkan sebagai pengidentifikasi tabel aktual, bukan nama topik. Oleh karena itu, Anda harus mengonfigurasi strategi pemetaan antara nama tabel dan topik dengan cermat.

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 alih-alih 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, saat Anda menggunakan aturan rute untuk mengubah topik tujuan, hal ini juga memodifikasi nama tabel dalam pesan Kafka (dalam format Debezium JSON atau Canal JSON). Dalam kasus ini, 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. Hal ini memastikan bahwa sistem downstream dapat mengidentifikasi nama tabel sumber asli secara akurat.

Semantik EXACTLY_ONCE

• Konfigurasikan Tingkat Isolasi Konsumen

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

  • read_committed: Hanya membaca data yang 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 mengganggu konsistensi.

• Timeout Transaksi dan Kehilangan Data

  Saat memulihkan dari Checkpoint, Realtime Compute for Apache Flink hanya mempertimbangkan Transaksi yang dikomit sebelum Checkpoint tersebut dimulai. Jika durasi antara Kegagalan Pekerjaan dan dimulainya ulang melebihi Timeout Transaksi Kafka, Kafka secara otomatis membatalkan Transaksi yang terbuka, yang dapat menyebabkan Kehilangan Data.

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

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

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

• Kolam Produsen dan Checkpoint Bersamaan

  Mode EXACTLY_ONCE menggunakan Kolam Produsen Kafka dengan ukuran tetap. Setiap checkpoint menggunakan satu produsen dari kolam ini. Jika jumlah checkpoint bersamaan melebihi ukuran kolam, Pekerjaan gagal.

  Konfigurasikan ukuran kolam produsen berdasarkan jumlah maksimum checkpoint bersamaan.

• Batasan Pengurangan Paralelisme

  Jika Pekerjaan gagal sebelum Checkpoint pertama selesai, informasi Kolam Produsen asli hilang saat dimulai ulang. Oleh karena itu, jangan Mengurangi Paralelisme Pekerjaan sebelum Checkpoint pertama selesai. Jika pengurangan 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 dimulai ulang atau Transaksi habis waktunya. 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 Jendela berbasis Waktu Event tidak menghasilkan output untuk Tabel Sumber Message Queue for Apache Kafka?

• Bagaimana cara menghubungkan ke Kluster Kafka yang diamankan?

• Bagaimana cara menyelesaikan konflik nama bidang?

• Ruang Kerja dan Namespace