Realtime Compute for Apache Flink：ApsaraMQ for Kafka

Informasi latar belakang

Prasyarat

Perhatian

Saat ini, penulisan transaksional tidak disarankan karena bug desain pada Flink dan Kafka. Saat Anda mengatur sink.delivery-guarantee = exactly-once, konektor Kafka mengaktifkan penulisan transaksional, dan terdapat tiga masalah yang diketahui:

• Setiap checkpoint menghasilkan ID transaksi. Jika interval checkpoint terlalu singkat, terlalu banyak ID transaksi yang dihasilkan. Koordinator kluster Kafka mungkin kehabisan memori, sehingga mengganggu stabilitas kluster Kafka.

• Setiap transaksi membuat instance produsen. Jika terlalu banyak transaksi dikomit secara bersamaan, Pengelola Tugas (TaskManager) mungkin kehabisan memori, sehingga mengganggu stabilitas pekerjaan Flink.

• Jika beberapa pekerjaan Flink menggunakan sink.transactional-id-prefix yang sama, ID transaksi yang dihasilkan mungkin bertabrakan. Jika satu pekerjaan gagal menulis, hal ini akan memblokir Log Start Offset (LSO) partisi Kafka agar tidak maju, yang memengaruhi semua konsumen yang membaca data dari partisi tersebut.

Jika Anda memerlukan semantik tepat-sekali, gunakan Upsert Kafka untuk menulis ke tabel kunci primer dan pastikan idempotensi dengan kunci primer tersebut. Jika Anda harus menggunakan penulisan transaksional, lihat Pertimbangan untuk semantik EXACTLY_ONCE.

Pemecahan masalah konektivitas jaringan

Jika pekerjaan Flink melaporkan error Timed out waiting for a node assignment saat startup, penyebabnya biasanya adalah masalah konektivitas jaringan antara Flink dan Kafka.

SQL

Konektor Kafka dapat digunakan dalam pekerjaan SQL sebagai tabel sumber atau tabel sink.

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 bidang `record_topic`.
  `record_topic` STRING NOT NULL METADATA FROM 'topic' VIRTUAL,
  -- Baca timestamp dari ConsumerRecord sebagai bidang `ts`.
  `ts` TIMESTAMP_LTZ(3) METADATA FROM 'timestamp' VIRTUAL,
  -- Baca offset pesan sebagai bidang `record_offset`.
  `record_offset` BIGINT NOT NULL METADATA FROM 'offset' VIRTUAL,
  ...
) WITH (
  'connector' = 'kafka',
  ...
);

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

Parameter WITH

• Umum

  Parameter	Deskripsi	Tipe data	Wajib	Nilai default	Keterangan
  connector	Jenis tabel.	String	Ya	Tidak ada	Nilainya tetap Kafka.
  properties.bootstrap.servers	Alamat broker Kafka.	String	Ya	Tidak ada	Formatnya host:port,host:port,host:port. Pisahkan alamat dengan koma (,).
  properties.*	Konfigurasi langsung untuk klien Kafka.	String	Tidak	Tidak ada	Akhiran harus merupakan konfigurasi yang didefinisikan dalam dokumentasi resmi Kafka untuk produsen dan konsumen. Flink menghapus awalan properties. dan meneruskan konfigurasi yang tersisa ke klien Kafka. Misalnya, Anda dapat menggunakan 'properties.allow.auto.create.topics'='false' untuk menonaktifkan pembuatan topik otomatis. Jangan gunakan metode ini untuk memodifikasi konfigurasi berikut, karena akan ditimpa oleh konektor Kafka: key.deserializer value.deserializer
  format	Format yang digunakan untuk membaca atau menulis bagian nilai pesan Kafka.	String	Tidak	Tidak ada	Format yang didukung: csv json avro debezium-json canal-json maxwell-json avro-confluent raw Catatan Untuk informasi selengkapnya tentang pengaturan parameter format, lihat Parameter Format.
  key.format	Format yang digunakan untuk membaca atau menulis bagian kunci pesan Kafka.	String	Tidak	Tidak ada	Format yang didukung: csv json avro debezium-json canal-json maxwell-json avro-confluent raw Catatan Jika Anda menggunakan konfigurasi ini, konfigurasi key.options wajib ditentukan.
  key.fields	Bidang di tabel sumber atau sink yang sesuai dengan bagian kunci pesan Kafka.	String	Tidak	Tidak ada	Pisahkan beberapa nama bidang dengan titik koma (;). Contohnya: field1;field2
  key.fields-prefix	Menentukan awalan khusus untuk semua bidang kunci pesan Kafka guna menghindari konflik nama dengan bidang di bagian nilai pesan.	String	Tidak	Tidak ada	Item konfigurasi ini hanya digunakan untuk membedakan nama kolom di tabel sumber dan sink. Awalan dihapus saat mengurai dan menghasilkan bagian kunci pesan Kafka. Catatan Jika Anda menggunakan konfigurasi ini, Anda harus mengatur value.fields-include ke EXCEPT_KEY.
  value.format	Format yang digunakan untuk membaca atau menulis bagian nilai pesan Kafka.	String	Tidak	Tidak ada	Konfigurasi ini setara dengan format. Anda hanya dapat mengatur salah satu dari format atau value.format. Jika keduanya dikonfigurasi, value.format akan menimpa format.
  value.fields-include	Menentukan apakah akan menyertakan bidang yang sesuai dengan bagian kunci pesan saat mengurai atau menghasilkan bagian nilai pesan Kafka.	String	Tidak	ALL	Nilai yang valid: ALL (default): Semua kolom diproses sebagai bagian nilai pesan Kafka. EXCEPT_KEY: Bidang yang tersisa, tidak termasuk yang didefinisikan di key.fields, diproses sebagai bagian nilai pesan Kafka.

• Tabel sumber

  Parameter	Deskripsi	Tipe data	Wajib	Nilai default	Keterangan
  topic	Nama topik untuk dibaca.	String	Tidak	Tidak ada	Pisahkan beberapa nama topik dengan titik koma (;), misalnya topic-1;topic-2. Catatan Anda hanya dapat menentukan salah satu opsi topic atau topic-pattern.
  topic-pattern	Ekspresi reguler yang cocok dengan nama topik untuk dibaca. Semua topik yang cocok dengan ekspresi reguler ini akan dibaca saat pekerjaan berjalan.	String	Tidak	Tidak ada	Catatan Anda hanya dapat menentukan salah satu opsi topic atau topic-pattern.
  properties.group.id	ID kelompok konsumen.	String	Tidak	KafkaSource-{source_table_name}	Jika ID grup yang ditentukan digunakan untuk pertama kalinya, Anda harus mengatur properties.auto.offset.reset ke earliest atau latest untuk menentukan offset awal.
  scan.startup.mode	Offset awal untuk membaca data dari Kafka.	String	Tidak	group-offsets	Nilai yang valid: earliest-offset: Mulai membaca dari partisi paling awal di Kafka. latest-offset: Mulai membaca dari offset terbaru di Kafka. group-offsets (default): Mulai membaca dari offset yang dikomit oleh properties.group.id yang ditentukan. timestamp: Mulai membaca dari timestamp yang ditentukan oleh scan.startup.timestamp-millis. specific-offsets: Mulai membaca dari offset yang ditentukan oleh scan.startup.specific-offsets. Catatan Parameter ini berlaku saat pekerjaan dimulai tanpa state. Saat pekerjaan dimulai ulang dari checkpoint atau pulih dari state, pekerjaan akan menggunakan progres yang disimpan dalam state untuk melanjutkan pembacaan.
  scan.startup.specific-offsets	Dalam mode startup specific-offsets, menentukan offset awal untuk setiap partisi.	String	Tidak	Tidak ada	Contoh: partition:0,offset:42;partition:1,offset:300
  scan.startup.timestamp-millis	Dalam mode startup timestamp, menentukan timestamp offset awal.	Long	Tidak	Tidak ada	Unitnya milidetik.
  scan.topic-partition-discovery.interval	Interval untuk menemukan topik dan partisi Kafka secara dinamis.	Duration	Tidak	5 menit	Interval pemeriksaan partisi default adalah 5 menit. Untuk menonaktifkan fitur ini, Anda harus secara eksplisit mengatur interval pemeriksaan partisi ke nilai non-positif. Saat penemuan partisi dinamis diaktifkan, sumber Kafka dapat secara otomatis menemukan partisi baru dan membaca data darinya. Dalam mode topic-pattern, sumber Kafka tidak hanya membaca data dari partisi baru topik yang ada, tetapi juga membaca data dari semua partisi topik baru yang cocok dengan ekspresi reguler. Catatan Dalam 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 data berdasarkan apakah data Kafka berisi header tertentu.	String	Tidak	Tidak ada	Pisahkan kunci header dan nilai dengan titik dua (:). Hubungkan beberapa kondisi header dengan operator logika (&, |). Operator logika NOT (!) juga didukung. Misalnya, depart:toy|depart:book&!env:test menyimpan data Kafka yang headernya berisi depart=toy atau depart=book, dan tidak berisi env=test. Catatan Parameter ini hanya didukung di Ververica Runtime (VVR) 8.0.6 dan versi lebih baru. Tanda kurung tidak didukung dalam operasi logika. Operasi logika dilakukan dari kiri ke kanan. Nilai header dikonversi ke string dalam format UTF-8 untuk perbandingan dengan nilai header yang ditentukan.
  scan.check.duplicated.group.id	Menentukan apakah akan memeriksa duplikasi kelompok konsumen yang ditentukan oleh properties.group.id.	Boolean	Tidak	false	Nilai yang valid: true: Sebelum memulai pekerjaan, sistem memeriksa duplikasi kelompok konsumen. Jika ditemukan duplikasi, pekerjaan melaporkan error dan berhenti, mencegah konflik dengan kelompok konsumen yang ada. false: Pekerjaan langsung dimulai tanpa memeriksa konflik kelompok konsumen. Catatan Parameter ini hanya didukung di VVR 6.0.4 dan versi lebih baru.

• Tabel sink

  Parameter	Deskripsi	Tipe data	Wajib	Nilai default	Keterangan
  topic	Nama topik untuk ditulis.	String	Ya	Tidak ada	Tidak ada
  sink.partitioner	Mode pemetaan dari konkurensi Flink ke partisi Kafka.	String	Tidak	default	Nilai yang valid: default: Menggunakan partitioner Kafka default. fixed: Setiap konkurensi Flink sesuai dengan partisi Kafka tetap. round-robin: Data dari konkurensi Flink dialokasikan ke partisi Kafka secara round-robin. Partitioner kustom: Jika fixed dan round-robin tidak memenuhi kebutuhan Anda, Anda dapat membuat kelas turunan dari FlinkKafkaPartitioner untuk mendefinisikan partitioner kustom. Contohnya: org.mycompany.MyPartitioner
  sink.delivery-guarantee	Semantik pengiriman untuk tabel sink Kafka.	String	Tidak	at-least-once	Nilai yang valid: none: Tanpa jaminan. Data mungkin hilang atau diduplikasi. at-least-once (default): Menjamin tidak ada data yang hilang, tetapi data mungkin diduplikasi. exactly-once: Menggunakan transaksi Kafka untuk menjamin data tidak hilang atau diduplikasi. Catatan Saat menggunakan semantik exactly-once, parameter sink.transactional-id-prefix wajib ditentukan.
  sink.transactional-id-prefix	Awalan untuk ID transaksi Kafka yang digunakan dalam semantik exactly-once.	String	Tidak	Tidak ada	Konfigurasi ini hanya berlaku saat sink.delivery-guarantee diatur ke exactly-once.
  sink.parallelism	Tingkat paralelisme untuk operator tabel sink Kafka.	Integer	Tidak	Tidak ada	Konkurensi operator hulu ditentukan oleh framework.

Keamanan dan autentikasi

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*/
  /*Konfigurasikan path ke truststore (sertifikat CA) yang disediakan oleh server.*/
  /*File yang diunggah melalui Manajemen File disimpan di path /flink/usrlib/.*/
  'properties.ssl.truststore.location' = '/flink/usrlib/kafka.client.truststore.jks',
  'properties.ssl.truststore.password' = 'test1234',
  /*Jika autentikasi klien diperlukan, konfigurasikan path ke keystore (kunci privat).*/
  'properties.ssl.keystore.location' = '/flink/usrlib/kafka.client.keystore.jks',
  'properties.ssl.keystore.password' = 'test1234',
  /*Algoritma bagi klien untuk memverifikasi alamat server. Nilai kosong menonaktifkan verifikasi alamat server.*/
  'properties.ssl.endpoint.identification.algorithm' = '',
  /*Konfigurasi SASL*/
  /*Atur mekanisme SASL ke SCRAM-SHA-256.*/
  'properties.sasl.mechanism' = 'SCRAM-SHA-256',
  /*Konfigurasikan JAAS*/
  'properties.sasl.jaas.config' = 'org.apache.flink.kafka.shaded.org.apache.kafka.common.security.scram.ScramLoginModule required username="username" password="password";'
)

Anda dapat mengunggah sertifikat CA dan kunci privat yang disebutkan dalam contoh ke platform menggunakan fitur Manajemen File di Konsol Realtime Compute. Setelah diunggah, file disimpan di direktori /flink/usrlib. Jika file sertifikat CA yang akan digunakan bernama my-truststore.jks, tentukan 'properties.ssl.truststore.location' = '/flink/usrlib/my-truststore.jks' dalam parameter WITH untuk menggunakan sertifikat ini.

Offset awal tabel sumber

CREATE TEMPORARY TABLE kafka_source (
  ...
) WITH (
  'connector' = 'kafka',
  ...
  -- Mulai mengonsumsi dari offset paling awal.
  'scan.startup.mode' = 'earliest-offset',
  -- Mulai mengonsumsi dari offset terbaru.
  'scan.startup.mode' = 'latest-offset',
  -- Mulai mengonsumsi 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 kalinya, mulai mengonsumsi dari offset paling awal.
  'properties.auto.offset.reset' = 'latest', -- Jika "my-group" digunakan untuk pertama kalinya, mulai mengonsumsi dari offset terbaru.
  -- Mulai mengonsumsi dari timestamp milidetik yang ditentukan 1655395200000.
  'scan.startup.mode' = 'timestamp',
  'scan.startup.timestamp-millis' = '1655395200000',
  -- Mulai mengonsumsi dari offset yang ditentukan.
  'scan.startup.mode' = 'specific-offsets',
  'scan.startup.specific-offsets' = 'partition:0,offset:42;partition:1,offset:300'
);

Komit offset tabel sumber

Tabel sumber Kafka hanya mengkomit offset konsumen saat ini ke kluster Kafka setelah checkpoint berhasil. Jika interval checkpoint panjang, offset konsumen yang diamati di kluster Kafka akan tertinggal. Selama checkpoint, tabel sumber Kafka menyimpan progres baca saat ini dalam statenya dan tidak bergantung pada offset yang dikomit ke kluster untuk pemulihan kesalahan. Pengkomitan offset hanya untuk memantau progres baca di sisi Kafka. Kegagalan komit offset tidak memengaruhi kebenaran data.

Partitioner kustom untuk tabel sink

Jika partitioner produsen Kafka bawaan tidak memenuhi kebutuhan Anda, Anda dapat mengimplementasikan partitioner kustom untuk menulis data ke partisi yang sesuai. Partitioner kustom harus mewarisi dari FlinkKafkaPartitioner. Setelah pengembangan, kompilasi paket JAR dan unggah ke Konsol Realtime Compute menggunakan fitur Manajemen File. Setelah diunggah dan direferensikan, atur parameter sink.partitioner dalam klausa WITH ke path kelas lengkap partitioner tersebut, seperti org.mycompany.MyPartitioner.

Memilih antara Kafka, Upsert Kafka, dan katalog Kafka JSON

Kafka adalah sistem antrian pesan append-only yang tidak mendukung pembaruan atau penghapusan data. Oleh karena itu, Kafka tidak dapat menangani data Change Data Capture (CDC) hulu atau logika penarikan dari operator seperti agregasi dan join dalam SQL streaming. Untuk menulis data dengan perubahan atau penarikan ke Kafka, gunakan tabel sink Upsert Kafka, yang dirancang khusus untuk menangani data perubahan.

Untuk menyinkronkan data perubahan dari satu atau beberapa tabel di database hulu ke Kafka secara batch dengan mudah, Anda dapat menggunakan katalog Kafka JSON. Jika data yang disimpan di Kafka dalam format JSON, penggunaan katalog Kafka JSON menghilangkan kebutuhan untuk mendefinisikan skema dan parameter WITH. Untuk informasi selengkapnya, 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 sumber Kafka

  Sumber Kafka menyediakan kelas builder untuk membuat instance KafkaSource. Contoh kode berikut menunjukkan cara membuat sumber Kafka untuk mengonsumsi data dari offset paling awal `input-topic`. Kelompok konsumen bernama `my-group`, dan isi pesan Kafka dideserialisasi menjadi 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");

  Saat membuat KafkaSource, Anda harus menentukan parameter berikut.

  Parameter	Deskripsi
  BootstrapServers	Alamat broker Kafka. Konfigurasikan ini menggunakan metode setBootstrapServers(String).
  GroupId	ID kelompok konsumen. Konfigurasikan ini menggunakan metode setGroupId(String).
  Topics or Partition	Nama topik atau partisi yang berlangganan. Sumber Kafka menyediakan tiga cara berikut untuk berlangganan topik atau partisi: Daftar topik: Berlangganan semua partisi dalam daftar topik. KafkaSource.builder().setTopics("topic-a","topic-b") Pencocokan ekspresi reguler: Berlangganan semua partisi topik yang cocok dengan ekspresi reguler. KafkaSource.builder().setTopicPattern("topic.*") Daftar partisi: Berlangganan partisi yang ditentukan. 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 untuk mengurai pesan Kafka. Tentukan deserializer menggunakan setDeserializer(KafkaRecordDeserializationSchema), di mana KafkaRecordDeserializationSchema mendefinisikan cara mengurai ConsumerRecord Kafka. Jika Anda hanya perlu mengurai data dalam isi pesan (nilai) pesan Kafka, Anda dapat melakukannya dengan salah satu cara berikut: Gunakan metode setValueOnlyDeserializer(DeserializationSchema) dalam kelas builder KafkaSource yang disediakan oleh Flink. DeserializationSchema mendefinisikan cara mengurai data biner dalam isi pesan Kafka. Gunakan parser yang disediakan oleh Kafka, yang mencakup beberapa kelas implementasi. Misalnya, Anda dapat menggunakan StringDeserializer untuk mengurai isi pesan Kafka menjadi string. import org.apache.kafka.common.serialization.StringDeserializer; KafkaSource.<String>builder() .setDeserializer(KafkaRecordDeserializationSchema.valueOnly(StringDeserializer.class)); Catatan Untuk mengurai ConsumerRecord sepenuhnya, Anda harus mengimplementasikan antarmuka KafkaRecordDeserializationSchema sendiri.

  XML

  Konektor DataStream Kafka tersedia di Repositori Maven Central.

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

  Saat menggunakan konektor DataStream Kafka, Anda perlu memahami properti Kafka berikut:

  • Offset awal konsumen

    Sumber Kafka dapat menentukan offset awal untuk konsumsi melalui inisialisator offset (OffsetsInitializer). Inisialisator offset bawaan meliputi yang berikut.

    Inisialisator offset	Pengaturan kode
    Mulai mengonsumsi dari offset paling awal.	KafkaSource.builder().setStartingOffsets(OffsetsInitializer.earliest())
    Mulai mengonsumsi dari offset terbaru.	KafkaSource.builder().setStartingOffsets(OffsetsInitializer.latest())
    Mulai mengonsumsi dari data dengan timestamp lebih besar dari atau sama dengan waktu yang ditentukan, dalam milidetik.	KafkaSource.builder().setStartingOffsets(OffsetsInitializer.timestamp(1592323200000L))
    Mulai mengonsumsi dari offset yang dikomit oleh kelompok konsumen. Jika offset yang dikomit tidak ada, gunakan offset paling awal.	KafkaSource.builder().setStartingOffsets(OffsetsInitializer.committedOffsets(OffsetResetStrategy.EARLIEST))
    Mulai mengonsumsi dari offset yang dikomit oleh kelompok konsumen, tanpa menentukan strategi pengaturan ulang offset.	KafkaSource.builder().setStartingOffsets(OffsetsInitializer.committedOffsets())

    Catatan

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

    • Jika tidak ada inisialisator offset yang ditentukan, OffsetsInitializer.earliest() (offset paling awal) digunakan secara default.

  • Mode streaming dan batch

    Sumber Kafka mendukung mode runtime streaming dan batch. Secara default, sumber Kafka diatur untuk berjalan dalam mode streaming, sehingga pekerjaan tidak pernah berhenti sampai pekerjaan Flink gagal atau dibatalkan. Untuk mengonfigurasi sumber Kafka agar berjalan dalam mode batch, Anda dapat menggunakan setBounded(OffsetsInitializer) untuk menentukan offset berhenti. Saat semua partisi mencapai offset berhentinya, sumber Kafka keluar.

    Catatan

    Umumnya, tidak ada offset berhenti dalam mode streaming. Untuk memudahkan debugging kode, Anda dapat menggunakan setUnbounded(OffsetsInitializer) untuk menentukan offset berhenti dalam mode streaming. Perhatikan bahwa nama metode untuk menentukan offset berhenti dalam mode streaming dan batch (setUnbounded dan setBounded) berbeda.

  • Penemuan partisi dinamis

    Untuk menangani skenario seperti penskalaan topik atau pembuatan topik baru tanpa memulai ulang pekerjaan Flink, Anda dapat mengaktifkan fitur penemuan partisi dinamis dalam mode langganan topik atau partisi yang disediakan.

    Catatan

    Fitur penemuan partisi dinamis diaktifkan secara default, dan interval pemeriksaan partisi adalah 5 menit. Untuk menonaktifkan fitur ini, Anda harus secara eksplisit mengatur interval pemeriksaan partisi ke nilai non-positif. Kode berikut memberikan contoh.

    KafkaSource.builder()
        .setProperty("partition.discovery.interval.ms", "10000") // Periksa 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 situasi aktual.

  • Waktu event dan watermark

    Secara default, sumber Kafka menggunakan timestamp dalam pesan Kafka sebagai waktu event. Anda dapat mendefinisikan strategi watermark kustom untuk mengekstrak waktu event dari pesan dan mengirim watermark ke downstream.

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

    Untuk informasi selengkapnya tentang strategi watermark kustom, lihat Generating Watermarks.

    Catatan

    Jika beberapa tugas sumber paralel menganggur untuk waktu yang lama (misalnya, jika partisi Kafka tidak memiliki input data untuk waktu yang lama, atau jika konkurensi sumber melebihi jumlah partisi Kafka), mekanisme pembuatan watermark mungkin gagal. Dalam kasus ini, sistem tidak dapat memicu perhitungan window secara normal, yang menyebabkan aliran pemrosesan data macet.

    Untuk mengatasi masalah ini, Anda dapat melakukan penyesuaian berikut:

    • Konfigurasikan mekanisme timeout watermark: Aktifkan parameter table.exec.source.idle-timeout untuk memaksa sistem menghasilkan watermark setelah periode timeout tertentu. Ini memastikan siklus perhitungan window berjalan.

    • Optimalkan sumber data: Kami merekomendasikan mempertahankan rasio partisi Kafka terhadap konkurensi sumber yang wajar (disarankan: jumlah partisi ≥ tingkat paralelisme sumber).

  • Consumer Offset Commit

    Sumber Kafka mengkomit offset konsumen saat ini ketika checkpoint selesai. Ini memastikan bahwa state checkpoint Flink konsisten dengan offset yang dikomit di broker Kafka. Jika checkpointing tidak diaktifkan, sumber Kafka bergantung pada logika komit offset otomatis internal konsumen Kafka. Fitur komit otomatis dikonfigurasi oleh item konfigurasi konsumen Kafka enable.auto.commit dan auto.commit.interval.ms.

    Catatan

    Sumber Kafka tidak bergantung pada offset yang dikomit di broker untuk memulihkan pekerjaan yang gagal. Pengkomitan offset hanya untuk melaporkan progres konsumsi konsumen Kafka dan kelompok konsumen untuk pemantauan di sisi broker.

  • Properti lainnya

    Selain properti yang disebutkan di atas, Anda dapat menggunakan setProperties(Properties) dan setProperty(String, String) untuk mengatur properti apa pun untuk sumber Kafka dan konsumen Kafka. KafkaSource biasanya memiliki item konfigurasi berikut.

    Item konfigurasi	Deskripsi
    client.id.prefix	Menentukan awalan ID klien untuk konsumen Kafka.
    partition.discovery.interval.ms	Menentukan interval di mana sumber Kafka memeriksa partisi baru. Catatan Nilai partition.discovery.interval.ms ditimpa menjadi -1 dalam mode Batch.
    register.consumer.metrics	Menentukan apakah akan mendaftarkan metrik konsumen Kafka di Flink.
    Konfigurasi konsumen Kafka lainnya	Untuk informasi selengkapnya tentang konfigurasi konsumen Kafka, lihat Apache Kafka. Penting Konektor Kafka secara paksa menimpa beberapa parameter yang dikonfigurasi secara manual sebagai berikut: key.deserializer selalu ditimpa ke ByteArrayDeserializer. value.deserializer selalu ditimpa ke ByteArrayDeserializer. auto.offset.reset.strategy ditimpa ke OffsetsInitializer#getAutoOffsetResetStrategy().

    Contoh berikut menunjukkan cara mengonfigurasi konsumen Kafka untuk menggunakan PLAIN sebagai mekanisme SASL 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

    Sumber Kafka mendaftarkan metrik di Flink untuk pemantauan dan diagnostik.

    • Cakupan metrik

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

      Misalnya, offset konsumen saat ini (currentOffset) untuk partisi 1 topik "my-topic" didaftarkan di <some_parent_groups>.operator.KafkaSourceReader.topic.my-topic.partition.1.currentOffset. Jumlah komit offset yang berhasil (commitsSucceeded) didaftarkan di <some_parent_groups>.operator.KafkaSourceReader.commitsSucceeded.

    • Daftar metrik

      Nama metrik	Deskripsi	Cakupan
      currentOffset	Offset konsumen saat ini.	TopicPartition
      committedOffset	Offset yang dikomit saat ini.	TopicPartition
      commitsSucceeded	Jumlah komit yang berhasil.	KafkaSourceReader
      commitsFailed	Jumlah komit yang gagal.	KafkaSourceReader

    • Metrik konsumen Kafka

      Metrik konsumen Kafka didaftarkan di grup metrik KafkaSourceReader.KafkaConsumer. Misalnya, metrik konsumen Kafka records-consumed-total didaftarkan di <some_parent_groups>.operator.KafkaSourceReader.KafkaConsumer.records-consumed-total.

      Anda dapat menggunakan item konfigurasi register.consumer.metrics untuk menentukan apakah akan mendaftarkan metrik konsumen Kafka. Secara default, opsi ini diatur ke true. Untuk informasi selengkapnya tentang metrik konsumen Kafka, lihat Apache Kafka.

• Buat sink Kafka

  Sink Kafka Flink dapat menulis data stream ke satu atau beberapa topik Kafka.

  DataStream<String> stream = ...
  
  
  Properties properties = new Properties();
  properties.setProperty("bootstrap.servers", );
  KafkaSink<String> kafkaSink =
                  KafkaSink.<String>builder()
                          .setKafkaProducerConfig(kafkaProperties) // konfigurasi produsen
                          .setRecordSerializer(
                                  KafkaRecordSerializationSchema.builder()
                                          .setTopic("my-topic") // topik target
                                          .setKafkaValueSerializer(StringSerializer.class) // skema serialisasi
                                          .build())
                          .setDeliveryGuarantee(DeliveryGuarantee.EXACTLY_ONCE) // toleransi kesalahan
                          .build();
  
  stream.sinkTo(kafkaSink);

  Anda perlu mengonfigurasi parameter berikut.

  Parameter	Deskripsi
  Topik	Nama topik default tempat data ditulis.
  Serialisasi data	Saat membangun, Anda perlu menyediakan KafkaRecordSerializationSchema untuk mengonversi data input menjadi ProducerRecord Kafka. Flink menyediakan builder skema untuk menawarkan beberapa komponen umum, seperti serialisasi kunci/nilai pesan, pemilihan topik, dan partisi pesan. Anda juga dapat mengimplementasikan antarmuka yang sesuai untuk kontrol lebih lanjut. Metode ProducerRecord<byte[], byte[]> serialize(T element, KafkaSinkContext context, Long timestamp) dipanggil untuk setiap catatan data yang masuk untuk menghasilkan ProducerRecord yang akan ditulis ke Kafka. Anda dapat memiliki kontrol detail halus atas cara setiap catatan data ditulis ke Kafka. Dengan ProducerRecord, Anda dapat melakukan operasi berikut: Atur nama topik tempat menulis. Tentukan kunci pesan. Tentukan partisi tempat data ditulis.
  Properti klien Kafka	bootstrap.servers wajib diisi. Ini adalah daftar broker Kafka yang dipisahkan koma.
  Semantik toleransi kesalahan	Saat checkpointing Flink diaktifkan, sink Kafka Flink dapat menjamin semantik tepat-sekali. Selain mengaktifkan checkpointing Flink, Anda juga dapat menentukan semantik toleransi kesalahan yang berbeda melalui parameter DeliveryGuarantee. Detail parameter DeliveryGuarantee adalah sebagai berikut: DeliveryGuarantee.NONE: (Pengaturan default) Flink tidak memberikan jaminan apa pun. Data mungkin hilang atau diduplikasi. DeliveryGuarantee.AT_LEAST_ONCE: Menjamin tidak ada data yang hilang, tetapi data mungkin diduplikasi. DeliveryGuarantee.EXACTLY_ONCE: Menggunakan transaksi Kafka untuk menyediakan semantik tepat-sekali. Catatan Untuk pertimbangan saat menggunakan semantik EXACTLY_ONCE, lihat Pertimbangan untuk semantik EXACTLY_ONCE.

Strategi pemetaan nama tabel ke topik

Saat menggunakan Kafka sebagai target untuk pekerjaan integrasi data, Anda perlu mengonfigurasi strategi pemetaan nama tabel ke topik dengan hati-hati. Hal ini karena format pesan Kafka yang ditulis (Debezium JSON atau Canal JSON) juga berisi informasi nama tabel, dan konsumsi pesan Kafka selanjutnya sering menggunakan informasi nama tabel dalam data sebagai nama tabel aktual (bukan nama topik).

Anggaplah Anda perlu menyinkronkan dua tabel, `mydb.mytable1` dan `mydb.mytable2`, dari MySQL. Strategi konfigurasi yang mungkin adalah sebagai berikut:

1. Jangan mengonfigurasi strategi pemetaan apa pun

Tanpa strategi pemetaan apa pun, setiap tabel ditulis ke topik dengan 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`. Kode berikut memberikan 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. Konfigurasikan aturan route untuk pemetaan (tidak disarankan)

Dalam banyak skenario, pengguna tidak ingin topik yang ditulis berada dalam format "nama_database.nama_tabel". Mereka ingin menulis data ke topik tertentu. Oleh karena itu, mereka mengonfigurasi aturan route untuk pemetaan. Kode berikut memberikan 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: mytable1

Dalam kasus ini, semua data dari `mydb.mytable1` dan `mydb.mytable2` ditulis ke topik `mytable1`.

Namun, saat Anda menggunakan aturan route untuk mengubah nama topik yang ditulis, hal ini juga memodifikasi informasi nama tabel dalam pesan Kafka (format Debezium JSON atau Canal JSON). Dalam kasus ini, semua nama tabel dalam pesan Kafka menjadi `mytable1`. Hal ini dapat menyebabkan perilaku yang tidak diinginkan saat sistem lain mengonsumsi pesan Kafka dari topik ini.

3. Konfigurasikan parameter sink.tableId-to-topic.mapping untuk pemetaan (disarankan)

Untuk mengonfigurasi aturan pemetaan nama tabel ke topik sambil mempertahankan informasi nama tabel sumber, gunakan parameter `sink.tableId-to-topic.mapping`. Kode berikut memberikan 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

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, semua data dari `mydb.mytable1` dan `mydb.mytable2` ditulis ke topik `mytable1`. Informasi nama tabel dalam pesan Kafka (format Debezium JSON atau Canal JSON) tetap `mydb.mytable1` atau `mydb.mytable2`. Saat sistem lain mengonsumsi pesan Kafka dari topik ini, mereka dapat memperoleh informasi nama tabel sumber dengan benar.

Pertimbangan untuk semantik EXACTLY_ONCE

• Konfigurasikan tingkat isolasi konsumen

  Semua aplikasi yang mengonsumsi data Kafka harus mengatur `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, yang merusak konsistensi.

• Timeout transaksi dan kehilangan data

  Saat Flink pulih dari checkpoint, Flink hanya bergantung pada transaksi yang telah dikomit sebelum checkpoint tersebut dimulai. Jika waktu antara kegagalan pekerjaan dan restart-nya melebihi timeout transaksi Kafka, Kafka secara otomatis membatalkan transaksi tersebut, menyebabkan kehilangan data.

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

  • transaction.timeout.ms default sink Kafka Flink adalah 1 jam.

  • Anda harus meningkatkan transaction.max.timeout.ms di sisi broker agar tidak kurang dari pengaturan Flink.

• Kolam produsen dan checkpoint konkuren

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

  Sesuaikan ukuran kolam produsen berdasarkan jumlah maksimum checkpoint konkuren.

• Batasan penskalaan turun tingkat paralelisme

  Jika pekerjaan gagal sebelum checkpoint pertama, informasi kolam produsen asli tidak dipertahankan saat restart. Oleh karena itu, jangan mengurangi tingkat paralelisme pekerjaan sebelum checkpoint pertama selesai. Jika Anda harus menskala turun, tingkat paralelisme tidak boleh lebih rendah dari FlinkKafkaProducer.SAFE_SCALE_DOWN_FACTOR.

• Pembacaan pemblokiran transaksi

  Dalam mode read_committed, transaksi terbuka apa pun (tidak dikomit maupun dibatalkan) memblokir pembacaan dari seluruh topik.

  Misalnya:

  • Transaksi 1 menulis data.

  • Transaksi 2 menulis dan mengkomit data.

  • Selama Transaksi 1 belum selesai, data dari Transaksi 2 tidak terlihat oleh konsumen.

  Oleh karena itu:

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

  • Saat pekerjaan gagal, topik yang sedang ditulis akan memblokir konsumen sampai pekerjaan dijalankan ulang atau transaksi timeout. Dalam kasus ekstrem, timeout transaksi bahkan dapat memengaruhi pembacaan.

FAQ

• Bagaimana Flink mengambil data JSON dari Kafka?

• Mengapa tabel sumber Kafka dengan window berbasis waktu event tidak mengeluarkan data?

• Bagaimana cara menghubungkan ke kluster Kafka dengan informasi keamanan yang dikonfigurasi?

• Bagaimana cara mengatasi konflik penamaan bidang?

• Manajemen dan operasi penyimpanan