Realtime Compute for Apache Flink:Kafka connector

Ikhtisar

Prasyarat

Catatan

Penulisan transaksional tidak disarankan karena 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-masalah 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 instance 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 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',
  ...
);

Opsi konektor

• Umum

  Opsi	Deskripsi	Tipe	Wajib	Default	Keterangan
  connector	Jenis connector.	String	Ya	–	Nilainya harus kafka.
  properties.bootstrap.servers	Daftar alamat broker Kafka.	String	Ya	–	Format: host1:port1,host2:port2,.... Pisahkan alamat dengan koma (,).
  properties.*	Properti tambahan untuk klien Kafka.	String	Tidak	–	Kunci properti harus merupakan opsi valid yang ditentukan 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. 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 Anda menggunakan konfigurasi ini, key.options wajib diisi.
  key.fields	Field dari skema tabel yang akan digunakan sebagai kunci pesan Kafka.	String	Tidak	–	Pisahkan beberapa nama field dengan titik koma (;). Misalnya, 'field1;field2'.
  key.fields-prefix	Awalan kustom untuk semua field kunci guna mencegah konflik nama dengan field nilai.	String	Tidak	–	Awalan ini digunakan untuk membedakan antara field kunci dan field 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 menggantikan format.
  value.fields-include	Menentukan apakah field 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 ditentukan dalam key.fields.

• Tabel sumber

  Opsi	Deskripsi	Tipe	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 salah satu opsi ini atau topic-pattern, tetapi tidak keduanya.
  topic-pattern	Ekspresi reguler yang cocok dengan topik yang akan dilanggan. 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 salah satu opsi ini atau topic, tetapi tidak keduanya.
  properties.group.id	ID kelompok konsumen 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 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 di-commit 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 tanpa state. Jika pekerjaan dilanjutkan dari checkpoint, pekerjaan membaca dari offset yang disimpan dalam state checkpoint.
  scan.startup.specific-offsets	Offset awal per-partisi saat scan.startup.mode adalah 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 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	–	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 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	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 dan versi setelahnya.

• Tabel sink

  Opsi	Deskripsi	Tipe	Wajib	Default	Keterangan
  topic	Topik tujuan.	String	Ya	–	–
  sink.partitioner	Memetakan record 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: 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. Record dapat 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 adalah exactly-once.	String	Ya, jika sink.delivery-guarantee adalah 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 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',
  ...
  -- Mengonsumsi dari offset paling awal.
  'scan.startup.mode' = 'earliest-offset',
  -- Mengonsumsi dari offset terbaru.
  'scan.startup.mode' = 'latest-offset',
  -- Mengonsumsi dari offset yang telah di-commit 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.
  -- Mengonsumsi dari timestamp tertentu dalam milidetik: 1655395200000.
  'scan.startup.mode' = 'timestamp',
  'scan.startup.timestamp-millis' = '1655395200000',
  -- Mengonsumsi dari offset tertentu.
  'scan.startup.mode' = 'specific-offsets',
  'scan.startup.specific-offsets' = 'partition:0,offset:42;partition:1,offset:300'
);

Menyimpan offset sumber

Tabel sumber Kafka hanya melakukan commit offset konsumennya ke kluster Kafka setelah checkpoint berhasil, sehingga interval checkpoint yang panjang menyebabkan offset yang di-commit tertinggal. Tabel sumber menyimpan progres bacaan aktual dalam state checkpoint, yang digunakan sistem untuk pemulihan kesalahan. Offset yang di-commit hanya berfungsi sebagai monitor progres dan tidak digunakan untuk pemulihan, sehingga kegagalan commit tidak memengaruhi akurasi data.

Pemartisi 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 dirujuk, atur parameter sink.partitioner di klausa WITH ke nama kelas lengkap partitioner Anda, misalnya, org.mycompany.MyPartitioner.

Kafka, Upsert Kafka, dan katalog JSON Kafka

Kafka adalah platform streaming event yang hanya menambahkan data dan tidak mendukung pembaruan atau penghapusan data. Di SQL streaming, tabel sink Kafka standar tidak dapat menangani data Change Data Capture (CDC) hulu atau logika retraction dari operator seperti agregat 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 JSON Kafka. Jika data yang disimpan di Kafka dalam format JSON, katalog JSON Kafka memungkinkan Anda melewati langkah mendefinisikan skema dan parameter WITH. Untuk detailnya, lihat Kelola katalog JSON Kafka.

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

• Membuat sumber Kafka

  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 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 telah di-commit Kelompok Konsumen. Jika tidak ada Offset yang telah di-commit, menggunakan strategi reset yang ditentukan (misalnya, Offset paling awal).	KafkaSource.builder().setStartingOffsets(OffsetsInitializer.committedOffsets(OffsetResetStrategy.EARLIEST))
    Konsumsi dimulai dari offset yang di-commit 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, 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 batas hingga gagal atau dibatalkan. Untuk mengonfigurasi Kafka Source agar berjalan dalam Mode Batch, Anda dapat menggunakan setBounded(OffsetsInitializer) untuk menentukan offset berhenti. Kafka Source berhenti 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 nama metode yang berbeda 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 me-restart 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 event dan watermark

    Secara default, Kafka Source menggunakan timestamp dari Pesan Kafka sebagai Waktu Event. Anda dapat mendefinisikan strategi Watermark kustom 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 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, sehingga dapat menghambat komputasi jendela di hilir.

    Untuk mengatasi masalah ini, pertimbangkan solusi berikut:

    • Konfigurasi timeout idle sumber: Aktifkan properti table.exec.source.idle-timeout untuk menandai sumber idle sebagai sementara idle. Hal ini memungkinkan Watermark di hulu maju.

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

  • Offset commit

    Saat checkpoint diaktifkan, Kafka Source melakukan commit Offset konsumen saat ini ke Kafka ketika Checkpoint selesai. Hal ini memastikan state Checkpoint Flink konsisten dengan Offset yang di-commit 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 di-commit untuk toleransi kesalahan dan pemulihan. 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 Mode Batch, 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.

    • Ruang lingkup metrik

      Semua metrik untuk 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 subgrup KafkaSourceReader.topic.<topic_name>.partition.<partition_id>.

      Misalnya, metrik Offset Konsumen saat ini (currentOffset) untuk Partisi 1 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	Ruang Lingkup
      currentOffset	Offset konsumen saat ini dari partisi.	TopicPartition
      committedOffset	Offset terakhir yang di-commit 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.

• Membuat sink Kafka

  Sink Kafka 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 rekaman	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 yang masuk untuk menghasilkan ProducerRecord yang akan ditulis ke Kafka. ProducerRecord memberikan kontrol rinci tentang cara setiap record ditulis ke Kafka, memungkinkan Anda untuk: Menetapkan Topik tujuan. Atur Message Key. Tentukan Partition 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. Data dapat hilang atau terduplikasi. 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
  # Enable Dirty Data Tolerance
  ingestion.ignore-errors: true
  # Tolerate up to 1000 dirty data records
  ingestion.error-tolerance.max-count: 1000

Dengan konfigurasi ini, pekerjaan terus berjalan selama tidak menemui lebih dari 1.000 record kotor. Setelah jumlah tersebut terlampaui, 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
  # Enable Dirty Data Tolerance
  ingestion.ignore-errors: true
  # Tolerate all dirty data records
  ingestion.error-tolerance.max-count: -1

Meskipun toleransi data kotor membuat pekerjaan Anda tetap berjalan, Anda mungkin juga ingin memeriksa record yang bermasalah. Anda juga mungkin ingin menganalisis data kotor untuk meningkatkan produsen Kafka Anda. Seperti yang 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
  # Enable Dirty Data Tolerance
  ingestion.ignore-errors: true
  # Tolerate all dirty data records
  ingestion.error-tolerance.max-count: -1

pipeline:
  dirty-data.collector:
    # Write dirty data to the TaskManager log file
    type: logger

Pemetaan nama tabel dan topik

Saat Kafka berfungsi sebagai sink Flink CDC, format paket (seperti Debezium JSON atau Canal JSON) menyematkan nama tabel asli. Konsumen hilir umumnya menggunakan nama yang disematkan tersebut sebagai pengidentifikasi tabel, bukan 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 direkomendasikan)

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 sumber tabel 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 Paket Kafka (dalam format Debezium JSON atau Canal JSON) tetap dipertahankan sebagai mydb.mytable1 atau mydb.mytable2. Dengan demikian, sistem downstream tetap dapat mengidentifikasi tabel sumber asli setiap Catatan.

Contoh konfigurasi

Contoh berikut menunjukkan konfigurasi untuk kasus penggunaan umum.

source:
  type: kafka
  topic: customers
  properties.bootstrap.servers: localhost:9092
  properties.group.id: ${kafka.group.id}
  value.format: json
  # Optional: Dynamically infer the schema of each record and detect schema changes.
  schema.inference.strategy: continuous

sink:
  type: paimon
  name: Paimon Sink
  catalog.properties.metastore: rest
  catalog.properties.uri: dlf_uri
  catalog.properties.warehouse: your_warehouse
  catalog.properties.token.provider: dlf
  # Optional: Specify a commit user. We recommend that you use a different commit user for each job to prevent conflicts.
  commit.user: your_job_name
  # Optional: Enable deletion vectors to improve read performance.
  table.properties.deletion-vectors.enabled: true

source:
  type: kafka
  topic-pattern: user_event_.*
  properties.bootstrap.servers: localhost:9092
  properties.group.id: ${kafka.group.id}
  value.format: json
  # Optional: Dynamically infer the schema of each record and detect schema changes.
  schema.inference.strategy: continuous

sink:
  type: starrocks
  jdbc-url: jdbc:mysql://<yourFeHostname>:9030
  load-url: <yourFeHostname>:8030
  username: <yourUsername>
  password: ${secret_values.starrocks_password}
 
  # Optional: For jobs with low data volumes, use a shorter flush interval to prevent data from remaining unwritten for a long time. The default is 300000 milliseconds, or 5 minutes.
  sink.buffer-flush.interval-ms: 5000
  # Optional: If the upstream character set is utf8mb4, set this parameter to 4 to prevent text truncation. The default is 3.
  unicode-char.max-bytes: 4
  # Optional: Specify the number of buckets for automatically created tables. You must explicitly configure this parameter for StarRocks versions earlier than 2.5.7. Later versions can infer the value.
  table.create.num-buckets: 8
  # Optional: Specify the number of replicas for automatically created tables based on your cluster configuration.
  table.create.properties.replication_num: 3
  # Optional: For StarRocks 3.2 and later, enable this feature to accelerate schema changes.
  table.create.properties.fast_schema_evolution: true
  # Note: If a transform changes a primary key, you must also set sink.ignore.update-before to false.
  # Otherwise, the row that uses the old primary key remains in the downstream system.

source:
  type: kafka
  topic: customers
  properties.bootstrap.servers: localhost:9092
  properties.group.id: ${kafka.group.id}
  value.format: json
  # Optional: Dynamically infer the schema of each record and detect schema changes.
  schema.inference.strategy: continuous
  metadata.list: topic,partition

sink:
  type: paimon
  name: Paimon Sink
  catalog.properties.metastore: rest
  catalog.properties.uri: dlf_uri
  catalog.properties.warehouse: your_warehouse
  catalog.properties.token.provider: dlf
  # Optional: Specify a commit user. We recommend that you use a different commit user for each job to prevent conflicts.
  commit.user: your_job_name
  # Optional: Enable deletion vectors to improve read performance.
  table.properties.deletion-vectors.enabled: true

source:
  type: kafka
  topic: customers
  properties.bootstrap.servers: localhost:9092
  properties.group.id: ${kafka.group.id}
  value.format: json
  # Optional: Dynamically infer the schema of each record and detect schema changes.
  schema.inference.strategy: continuous
  # Ignore parsing errors. By default, all parsing errors are ignored when this feature is enabled.
  ingestion.ignore-errors: true

sink:
  type: paimon
  name: Paimon Sink
  catalog.properties.metastore: rest
  catalog.properties.uri: dlf_uri
  catalog.properties.warehouse: your_warehouse
  catalog.properties.token.provider: dlf
  # Optional: Specify a commit user. We recommend that you use a different commit user for each job to prevent conflicts.
  commit.user: your_job_name
  # Optional: Enable deletion vectors to improve read performance.
  table.properties.deletion-vectors.enabled: true
  
# Enable a dirty data collector to print records that cannot be parsed.
pipeline:
  dirty-data.collector:
    name: Logger Dirty Data Collector
    type: logger

source:
  type: kafka
  topic: customers
  properties.bootstrap.servers: localhost:9092
  properties.group.id: ${kafka.group.id}
  value.format: json
  # Optional: Dynamically infer the schema of each record and detect schema changes.
  schema.inference.strategy: continuous
  # Ignore parsing errors.
  ingestion.ignore-errors: true
  # Fail the job after 30 parsing errors.
  ingestion.error-tolerance.max-count: 30
  
sink:
  type: paimon
  name: Paimon Sink
  catalog.properties.metastore: rest
  catalog.properties.uri: dlf_uri
  catalog.properties.warehouse: your_warehouse
  catalog.properties.token.provider: dlf
  # Optional: Specify a commit user. We recommend that you use a different commit user for each job to prevent conflicts.
  commit.user: your_job_name
  # Optional: Enable deletion vectors to improve read performance.
  table.properties.deletion-vectors.enabled: true
  
# Enable a dirty data collector to print records that cannot be parsed.
pipeline:
  dirty-data.collector:
    name: Logger Dirty Data Collector
    type: logger

source:
  type: kafka
  topic: customers
  properties.bootstrap.servers: localhost:9092
  properties.group.id: ${kafka.group.id}
  value.format: json
  # Optional: Dynamically infer the schema of each record and detect schema changes.
  schema.inference.strategy: continuous
  # Use the db and tbl fields as the table ID.
  value.json.decode.parser-table-id.fields: db,tbl

sink:
  type: paimon
  name: Paimon Sink
  catalog.properties.metastore: rest
  catalog.properties.uri: dlf_uri
  catalog.properties.warehouse: your_warehouse
  catalog.properties.token.provider: dlf
  # Optional: Specify a commit user. We recommend that you use a different commit user for each job to prevent conflicts.
  commit.user: your_job_name
  # Optional: Enable deletion vectors to improve read performance.
  table.properties.deletion-vectors.enabled: true
  
# Enable a dirty data collector to print records that cannot be parsed.
pipeline:
  dirty-data.collector:
    name: Logger Dirty Data Collector
    type: logger

source:
  type: kafka
  topic: customers
  properties.bootstrap.servers: localhost:9092
  properties.group.id: ${kafka.group.id}
  value.format: json
  # Use the db and tbl fields as the table ID.
  value.json.decode.parser-table-id.fields: db,tbl
  # Specify fixed types for selected fields.
  value.json.infer-schema.fixed-types: 'db STRING, tbl STRING, id BIGINT, amount DECIMAL(18, 2)'
  # Continue to dynamically infer fields that are not declared.
  schema.inference.strategy: continuous

sink:
  type: paimon
  name: Paimon Sink
  catalog.properties.metastore: rest
  catalog.properties.uri: dlf_uri
  catalog.properties.warehouse: your_warehouse
  catalog.properties.token.provider: dlf
  # Optional: Specify a commit user. We recommend that you use a different commit user for each job to prevent conflicts.
  commit.user: your_job_name
  # Optional: Enable deletion vectors to improve read performance.
  table.properties.deletion-vectors.enabled: true
  
# Enable a dirty data collector to print records that cannot be parsed.
pipeline:
  dirty-data.collector:
    name: Logger Dirty Data Collector
    type: logger

source:
  type: kafka
  topic: customers
  properties.bootstrap.servers: localhost:9092
  properties.group.id: ${kafka.group.id}
  value.format: canal-json
  # Optional: Dynamically infer the schema of each record and detect schema changes.
  schema.inference.strategy: continuous

sink:
  type: paimon
  name: Paimon Sink
  catalog.properties.metastore: rest
  catalog.properties.uri: dlf_uri
  catalog.properties.warehouse: your_warehouse
  catalog.properties.token.provider: dlf
  # Optional: Specify a commit user. We recommend that you use a different commit user for each job to prevent conflicts.
  commit.user: your_job_name
  # Optional: Enable deletion vectors to improve read performance.
  table.properties.deletion-vectors.enabled: true
  
# Enable a dirty data collector to print records that cannot be parsed.
pipeline:
  dirty-data.collector:
    name: Logger Dirty Data Collector
    type: logger

source:
  type: kafka
  topic: customers
  properties.bootstrap.servers: localhost:9092
  properties.group.id: ${kafka.group.id}
  value.format: canal-json
  # Optional: Dynamically infer the schema of each record and detect schema changes.
  schema.inference.strategy: continuous
  # Infer the schema from MySQL type information. You can set this parameter to SQL_TYPE to use SQL type information instead.
  value.canal-json.infer-schema.strategy: MYSQL_TYPE

sink:
  type: paimon
  name: Paimon Sink
  catalog.properties.metastore: rest
  catalog.properties.uri: dlf_uri
  catalog.properties.warehouse: your_warehouse
  catalog.properties.token.provider: dlf
  # Optional: Specify a commit user. We recommend that you use a different commit user for each job to prevent conflicts.
  commit.user: your_job_name
  # Optional: Enable deletion vectors to improve read performance.
  table.properties.deletion-vectors.enabled: true
  
# Enable a dirty data collector to print records that cannot be parsed.
pipeline:
  dirty-data.collector:
    name: Logger Dirty Data Collector
    type: logger

source:
  type: kafka
  topic: customers
  properties.bootstrap.servers: localhost:9092
  properties.group.id: ${kafka.group.id}
  value.format: debezium-json
  # Optional: Dynamically infer the schema of each record and detect schema changes.
  schema.inference.strategy: continuous

sink:
  type: paimon
  name: Paimon Sink
  catalog.properties.metastore: rest
  catalog.properties.uri: dlf_uri
  catalog.properties.warehouse: your_warehouse
  catalog.properties.token.provider: dlf
  # Optional: Specify a commit user. We recommend that you use a different commit user for each job to prevent conflicts.
  commit.user: your_job_name
  # Optional: Enable deletion vectors to improve read performance.
  table.properties.deletion-vectors.enabled: true

source:
  type: mysql
  hostname: #{hostname}
  port: 3306
  username: #{username}
  password: #{password}
  tables: order_dw.\.*
  server-id: 28601-28604
  # Optional: Synchronize data from tables that are created during the incremental phase.
  scan.binlog.newly-added-table.enabled: true
  # Optional: Synchronize table and column comments.
  include-comments.enabled: true
  # Optional: Process unbounded chunks first to prevent potential TaskManager out-of-memory errors.
  scan.incremental.snapshot.unbounded-chunk-first.enabled: true
  # Optional: Enable parsing filters to accelerate reads.
  scan.only.deserialize.captured.tables.changelog.enabled: true
  # Add mysqlType, sqlType, sql, isDdl, and other metadata to the Canal JSON data.
  include-binlog-meta.enable: true
  
sink:
  type: kafka
  properties.bootstrap.servers: localhost:9092
  topic: order_dw_tables
  # Use the Canal JSON changelog format for Kafka values.
  value.format: canal-json
  # Specify the format used to serialize date and time data.
  value.canal-json.timestamp-format.standard: SQL
  # Write all data to Partition 0 to preserve binlog order.
  partition.strategy: all-to-zero

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 di-commit.

  • read_uncommitted (Default): Dapat membaca data yang belum di-commit.

  EXACTLY_ONCE bergantung pada read_committed. Jika tidak, konsumen mungkin melihat data yang belum di-commit, sehingga mengganggu konsistensi.

• Timeout transaksi dan kehilangan data

  Saat memulihkan dari checkpoint, Realtime Compute for Apache Flink hanya mempertimbangkan transaksi yang telah di-commit 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 meningkatkan 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 dengan ukuran 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 melakukan penskalaan turun 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 di-commit atau dibatalkan memblokir operasi baca pada seluruh topik.

  Misalnya:

  • Transaksi 1 menulis data.

  • Transaksi 2 menulis lebih banyak data dan di-commit.

  • Selama Transaksi 1 tetap terbuka, data dari Transaksi 2 yang telah di-commit 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 diblokir untuk konsumen sampai pekerjaan restart 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 Waktu Event tidak menghasilkan output untuk Tabel Sumber ApsaraMQ for Kafka?

• Bagaimana cara menghubungkan ke Kluster Kafka yang diamankan?

• Bagaimana cara menyelesaikan konflik nama field?

• Ruang Kerja dan Namespace