Realtime Compute for Apache Flink：Message Queue for Kafka

Informasi latar belakang

Prasyarat

Catatan penggunaan

Saat ini, penulisan transaksional tidak disarankan karena keterbatasan desain di komunitas Flink dan Kafka. Saat Anda mengatur sink.delivery-guarantee = exactly-once, Kafka connector mengaktifkan penulisan transaksional, yang memiliki tiga masalah yang diketahui:

• Setiap checkpoint menghasilkan ID transaksi baru. Jika interval checkpoint terlalu pendek, jumlah ID transaksi menjadi berlebihan. Akibatnya, koordinator Kafka mungkin kehabisan memori, sehingga mengganggu stabilitas kluster Kafka.

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

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

Jika Anda memerlukan semantik tepat-sekali, gunakan Upsert Kafka untuk menulis ke tabel dengan primary key dan mengandalkan primary key tersebut untuk memastikan idempotensi. Untuk menggunakan penulisan transaksional, lihat Catatan tentang Semantik Tepat-Sekali.

Pemecahan masalah konektivitas jaringan

Jika pekerjaan Flink melaporkan error Timed out waiting for a node assignment saat startup, hal ini biasanya menunjukkan adanya masalah konektivitas jaringan antara Flink dan Kafka.

SQL

Anda dapat menggunakan Kafka connector 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 tempat pesan tersebut berada sebagai nilai bidang record_topic.
  `record_topic` STRING NOT NULL METADATA FROM 'topic' VIRTUAL,
  -- Baca timestamp di ConsumerRecord sebagai nilai bidang ts.
  `ts` TIMESTAMP_LTZ(3) METADATA FROM 'timestamp' VIRTUAL,
  -- Baca offset pesan sebagai nilai bidang record_offset.
  `record_offset` BIGINT NOT NULL METADATA FROM 'offset' VIRTUAL,
  ...
) WITH (
  'connector' = 'kafka',
  ...
);

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

dengan parameter

• Umum

  parameter	Deskripsi	Tipe data	Wajib	Nilai default	Keterangan
  connector	Tipe tabel.	String	Ya	Tidak ada	Atur nilainya ke kafka.
  properties.bootstrap.servers	Alamat IP dan nomor port broker Kafka.	String	Ya	Tidak ada	Format: host:port,host:port,host:port. Pisahkan beberapa pasangan host:port dengan koma (,).
  properties.*	Opsi yang dikonfigurasi untuk klien Kafka.	String	Tidak	Tidak ada	Akhiran harus mematuhi aturan yang ditentukan dalam Producer Configs dan Consumer Configs. Flink menghapus awalan "properties." dan meneruskan konfigurasi yang tersisa ke klien Kafka. Misalnya, Anda dapat menonaktifkan pembuatan topik otomatis menggunakan 'properties.allow.auto.create.topics'='false'. Konfigurasi berikut tidak dapat dimodifikasi dengan cara ini karena Kafka connector menimpa konfigurasi tersebut: key.deserializer value.deserializer
  format	Format yang digunakan untuk membaca atau menulis bidang nilai pesan Kafka.	String	Tidak	Tidak ada nilai default.	Format yang didukung csv json avro debezium-json canal-json maxwell-json avro-confluent raw Catatan Untuk informasi selengkapnya tentang opsi format, lihat Opsi Format.
  key.format	Format yang digunakan untuk membaca atau menulis bidang kunci pesan Kafka.	String	Tidak	Tidak ada nilai default.	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 diisi.
  key.fields	Bidang kunci di tabel sumber atau tabel sink yang sesuai dengan bidang kunci pesan Kafka.	String	Tidak	Tidak ada	Pisahkan beberapa nama bidang dengan titik koma (;), seperti field1;field2.
  key.fields-prefix	Awalan kustom untuk semua bidang kunci dalam pesan Kafka. Anda dapat mengonfigurasi opsi ini untuk mencegah konflik nama dengan bidang nilai.	String	Tidak	Tidak ada	Opsi ini hanya digunakan untuk membedakan nama kolom tabel sumber dan tabel sink. Awalan dihapus dari nama kolom saat bidang kunci pesan Kafka diurai atau dihasilkan. Catatan Saat Anda menggunakan konfigurasi ini, Anda harus mengatur opsi value.fields-include ke EXCEPT_KEY.
  value.format	Format yang digunakan untuk membaca atau menulis bidang nilai pesan Kafka.	String	Tidak	Tidak ada	Konfigurasi ini setara dengan format, dan Anda hanya dapat mengonfigurasi salah satu dari format atau value.format. Jika Anda mengonfigurasi keduanya, value.format akan menimpa format.
  value.fields-include	Menentukan apakah kunci pesan yang sesuai disertakan saat mengurai atau menghasilkan nilai pesan Kafka.	String	Tidak	ALL	Nilai yang valid: ALL (default): Semua bidang diproses sebagai nilai pesan Kafka. EXCEPT_KEY: Semua bidang kecuali bidang yang ditentukan oleh opsi key.fields diproses sebagai nilai pesan Kafka.

• Tabel sumber

  Parameter	Deskripsi	Tipe data	Wajib	Nilai default	Keterangan
  topic	Nama topik tempat Anda ingin membaca data.	String	Tidak	Tidak ada	Pisahkan beberapa nama topik dengan titik koma (;), seperti topic-1 dan topic-2. Catatan Anda tidak dapat menggunakan opsi topic bersamaan dengan opsi topic-pattern.
  topic-pattern	Ekspresi reguler yang digunakan untuk mencocokkan topik. Data dari semua topik yang namanya cocok dengan ekspresi reguler yang ditentukan akan dibaca saat deployment berjalan.	String	Tidak	Tidak ada	Catatan Anda tidak dapat menggunakan opsi topic bersamaan dengan opsi topic-pattern.
  properties.group.id	ID kelompok konsumen.	String	Tidak	KafkaSource-{Nama tabel sumber}	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 Kafka membaca data.	String	Tidak	group-offsets	Nilai yang valid: earliest-offset: Kafka membaca data dari partisi paling awal. latest-offset: Kafka membaca data dari offset terbaru. group-offsets (default): membaca data dari offset yang dikomit oleh kelompok konsumen dengan ID yang ditentukan oleh opsi properties.group.id. timestamp: Membaca data dari timestamp yang ditentukan oleh scan.startup.timestamp-millis. specific-offsets: membaca data dari offset yang ditentukan oleh opsi scan.startup.specific-offsets. Catatan Opsi ini berlaku saat deployment dimulai tanpa state. Saat deployment dimulai ulang dari checkpoint atau dilanjutkan dari state tertentu, deployment akan memprioritaskan membaca data dari progres yang disimpan dalam data state.
  scan.startup.specific-offsets	Offset awal setiap partisi saat opsi scan.startup.mode diatur ke specific-offsets.	String	Tidak	Tidak ada	Contoh: partition:0,offset:42;partition:1,offset:300.
  scan.startup.timestamp-millis	Timestamp offset awal saat opsi scan.startup.mode diatur ke timestamp.	Long	Tidak	Tidak ada	Unit: milidetik.
  scan.topic-partition-discovery.interval	Interval waktu untuk mendeteksi topik dan partisi Kafka secara dinamis.	Duration	Tidak	5 menit	Interval penemuan partisi default adalah 5 menit. Untuk menonaktifkan fitur ini, Anda harus secara eksplisit mengatur opsi ini ke nilai non-positif. Setelah fitur penemuan partisi dinamis diaktifkan, sumber Kafka dapat secara otomatis menemukan partisi baru dan membaca data dari partisi tersebut. Dalam mode topic-pattern, sumber Kafka membaca data dari partisi baru topik yang ada dan data dari semua partisi topik baru yang cocok dengan ekspresi reguler. Catatan Dalam Realtime Compute for Apache Flink yang menggunakan VVR 6.0.X, fitur penemuan partisi dinamis dinonaktifkan secara default. Dalam Realtime Compute for Apache Flink yang menggunakan VVR 8.0 atau versi lebih baru, fitur ini diaktifkan secara default. Interval penemuan partisi default adalah 5 menit.
  scan.header-filter	Penyaringan data Kafka berdasarkan apakah data tersebut mengandung header pesan tertentu.	String	Tidak	Tidak ada	Pisahkan kunci header dan nilai dengan titik dua (:). Pisahkan beberapa header dengan operator logika seperti AND (&) atau OR (|). Operator logika NOT (!) didukung. Misalnya, depart:toy|depart:book&!env:test menunjukkan bahwa data Kafka yang headernya mengandung depart=toy atau depart=book dan tidak mengandung env=test akan dipertahankan. Catatan Opsi ini hanya didukung untuk Realtime Compute for Apache Flink yang menggunakan VVR 8.0.6 atau versi lebih baru. Operasi tanda kurung tidak didukung. Operasi logika dilakukan dari kiri ke kanan secara berurutan. Nilai header dalam format UTF-8 dikonversi menjadi string dan dibandingkan dengan nilai header yang ditentukan oleh opsi scan.header-filter.
  scan.check.duplicated.group.id	Menentukan apakah akan memeriksa duplikasi kelompok konsumen yang ditentukan oleh parameter properties.group.id.	Boolean	Tidak	false	Nilai yang valid: true: Memeriksa duplikasi kelompok konsumen sebelum pekerjaan dimulai. Jika terdapat duplikasi kelompok konsumen, melaporkan error dan menangguhkan pekerjaan untuk mencegah konflik. false: Tidak memeriksa duplikasi kelompok konsumen sebelum pekerjaan dimulai. Catatan Opsi ini hanya didukung untuk Realtime Compute for Apache Flink yang menggunakan VVR 6.0.4 atau versi lebih baru.

• Khusus sink

  Parameter	Deskripsi	Tipe data	Wajib	Nilai default	Keterangan
  topic	Nama topik tempat data ditulis.	String	Ya	Tidak ada	N/A
  sink.partitioner	Pola untuk memetakan konkurensi Flink ke partisi Kafka.	String	Tidak	default	Nilai yang valid: default (default): Menggunakan partitioner Kafka default untuk mempartisi data. fixed: Setiap partisi Flink sesuai dengan partisi Kafka tetap. round-robin: Data dalam partisi Flink didistribusikan ke partisi Kafka secara round-robin. Pola pemetaan partisi kustom: Anda dapat membuat subclass FlinkKafkaPartitioner untuk mengonfigurasi pola pemetaan partisi kustom, seperti org.mycompany.MyPartitioner.
  sink.delivery-guarantee	Pola semantik untuk tabel sink Kafka.	String	Tidak	at-least-once	Nilai yang valid: none: Jaminan pengiriman tidak dipastikan. Data mungkin hilang atau diduplikasi. at-least-once (default): Memastikan data tidak hilang. Namun, data mungkin diduplikasi. exactly-once: Transaksi Kafka digunakan untuk memastikan semantik tepat-sekali. Ini memastikan data tidak hilang atau diduplikasi. Catatan Anda harus mengonfigurasi opsi sink.transactional-id-prefix jika Anda mengatur opsi ini ke exactly-once.
  sink.transactional-id-prefix	Awalan ID transaksi Kafka yang digunakan dalam semantik tepat-sekali.	String	Tidak	Tidak ada	Opsi ini hanya berlaku saat opsi sink.delivery-guarantee diatur ke exactly-once.
  sink.parallelism	Konkurensi operator untuk tabel sink Kafka.	Integer	Tidak	Tidak ada	Konkurensi operator hulu, yang ditentukan oleh framework.

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 Secure Sockets Layer (SSL).*/
  /*Tentukan jalur truststore sertifikat CA yang disediakan oleh server.*/
  /*Artefak yang diunggah disimpan di /flink/usrlib/.*/
  'properties.ssl.truststore.location' = '/flink/usrlib/kafka.client.truststore.jks',
  'properties.ssl.truststore.password' = 'test1234',
  /*Tentukan jalur file keystore kunci privat jika otentikasi klien diperlukan.*/
  'properties.ssl.keystore.location' = '/flink/usrlib/kafka.client.keystore.jks',
  'properties.ssl.keystore.password' = 'test1234',
  /*Algoritma yang digunakan klien untuk memverifikasi alamat server. Nilai null menunjukkan bahwa verifikasi alamat server dinonaktifkan.*/
  'properties.ssl.endpoint.identification.algorithm' = '',
  /*Konfigurasi SASL.*/
  /*Konfigurasi SCRAM-SHA-256 sebagai mekanisme SASL.*/
  '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 Artefak di konsol pengembangan untuk mengunggah sertifikat CA dan kunci privat dari contoh. Setelah diunggah, file tersebut disimpan di direktori /flink/usrlib. Jika file sertifikat CA yang ingin Anda gunakan bernama my-truststore.jks, Anda dapat mengatur parameter 'properties.ssl.truststore.location' dalam klausa WITH dengan dua cara berikut untuk menggunakan sertifikat ini:

• Jika Anda mengatur 'properties.ssl.truststore.location' = '/flink/usrlib/my-truststore.jks', Flink tidak perlu mengunduh file OSS secara dinamis selama runtime, tetapi mode debug tidak didukung.

• Untuk versi mesin komputasi waktu nyata Ververica Runtime (VVR) 11.5 dan versi lebih baru, Anda dapat mengonfigurasi properties.ssl.truststore.location dan properties.ssl.keystore.location dengan jalur mutlak OSS. Format jalur file adalah oss://flink-fullymanaged-<ID ruang kerja>/artifacts/namespaces/<nama proyek>/<nama file>. Metode ini mengunduh file OSS secara dinamis selama runtime Flink dan mendukung mode debug.

Offset awal untuk tabel sumber Kafka

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

Pengiriman Offset Tabel Sumber

Tabel sumber Kafka hanya mengirimkan offset konsumen ke kluster Kafka setelah operasi checkpoint berhasil. Jika interval checkpoint yang Anda tentukan terlalu besar, offset konsumen akan dikirimkan dengan penundaan ke kluster Kafka. Selama operasi checkpoint, tabel sumber Kafka menyimpan progres pembacaan data saat ini di backend status. Offset yang dikirimkan ke kluster Kafka tidak digunakan untuk pemulihan kesalahan. Offset yang dikirimkan hanya digunakan untuk memantau progres pembacaan data di Kafka. Akurasi data tidak terpengaruh bahkan jika offset gagal dikirimkan.

Partitioner kustom untuk tabel sink

Jika partitioner produsen Kafka bawaan tidak memenuhi kebutuhan Anda, Anda dapat mengimplementasikan partitioner kustom untuk menulis data ke partisi tertentu. Partitioner kustom harus mewarisi FlinkKafkaPartitioner. Setelah pengembangan, kompilasi paket JAR dan gunakan fitur Manajemen File untuk mengunggahnya ke konsol Komputasi Waktu Nyata. Setelah paket JAR diunggah dan direferensikan, atur parameter sink.partitioner dalam klausa WITH. Nilai parameter harus berupa jalur kelas lengkap partitioner, seperti org.mycompany.MyPartitioner.

Perbandingan antara Kafka, Upsert Kafka, dan katalog JSON Kafka

Kafka adalah sistem antrian pesan yang hanya mendukung penyisipan data dan tidak mendukung pembaruan atau penghapusan. Oleh karena itu, Kafka tidak dapat memproses data Change Data Capture (CDC) dari sistem hulu atau logika penarikan dari operator seperti agregat dan join selama komputasi SQL streaming. Jika Anda ingin menulis data yang mengandung data perubahan atau data penarikan ke Kafka, gunakan tabel sink Upsert Kafka, yang melakukan pemrosesan khusus pada data perubahan.

Jika Anda ingin menyinkronkan data perubahan dari satu atau beberapa tabel data di database hulu ke Kafka secara batch, Anda dapat menggunakan katalog JSON Kafka. Jika data yang disimpan di Kafka dalam format JSON, Anda dapat menggunakan katalog JSON Kafka. Hal ini menghilangkan kebutuhan untuk mengonfigurasi skema dan opsi dalam klausa WITH. Untuk informasi selengkapnya, 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. Perluas 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 sumber Kafka

  Sumber Kafka menyediakan kelas builder untuk membuat instance KafkaSource. Kode contoh berikut menunjukkan cara membuat sumber Kafka untuk mengonsumsi pesan dari offset paling awal topik "input-topic", dengan kelompok konsumen bernama my-group, dan mendeserialisasi isi pesan Kafka 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");

  Saat membuat KafkaSource, Anda harus menentukan parameter berikut.

  Parameter	Deskripsi
  BootstrapServers	Alamat broker Kafka. Anda dapat memanggil operasi setBootstrapServers(String) untuk mengonfigurasi alamat.
  GroupId	ID kelompok konsumen. Anda dapat memanggil metode setGroupId(String) untuk mengonfigurasi ID.
  Topik atau Partisi	Topik atau nama partisi tempat Anda berlangganan. Anda dapat mengonfigurasi sumber Kafka untuk berlangganan ke topik atau partisi menggunakan salah satu pola langganan berikut: Daftar topik. Setelah Anda mengonfigurasi daftar topik, sumber Kafka berlangganan ke semua partisi dari topik yang ditentukan. KafkaSource.builder().setTopics("topic-a","topic-b") Pola topik. Setelah Anda menentukan ekspresi reguler, sumber Kafka berlangganan ke semua partisi dari topik yang cocok dengan ekspresi reguler yang ditentukan. KafkaSource.builder().setTopicPattern("topic.*") Daftar partisi. Setelah Anda mengonfigurasi daftar partisi, sumber Kafka berlangganan ke 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 yang mendeserialisasi pesan Kafka. Anda dapat memanggil metode setDeserializer(KafkaRecordDeserializationSchema) untuk menentukan deserializer. Antarmuka KafkaRecordDeserializationSchema mendefinisikan cara objek ConsumerRecord dideserialisasi. Anda dapat menggunakan salah satu metode berikut untuk hanya mendeserialisasi bidang Value dalam pesan Kafka objek ConsumerRecord: Sumber Kafka menyediakan metode setValueOnlyDeserializer(DeserializationSchema). Kelas DeserializationSchema mendefinisikan cara pesan Kafka yang disimpan sebagai nilai biner dideserialisasi. Gunakan kelas yang mengimplementasikan Antarmuka Deserializer Kafka. Misalnya, Anda dapat menggunakan kelas StringDeserializer untuk mendeserialisasi pesan menjadi string. import org.apache.kafka.common.serialization.StringDeserializer; KafkaSource.<String>builder() .setDeserializer(KafkaRecordDeserializationSchema.valueOnly(StringDeserializer.class)); Catatan Jika Anda ingin mendeserialisasi objek ConsumerRecord, Anda harus membuat kelas yang mengimplementasikan antarmuka KafkaRecordDeserializationSchema.

  XML

  Konektor DataStream Kafka disimpan di repositori pusat Maven.

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

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

  • Offset awal

    Anda dapat menggunakan inisialisasi offset untuk menentukan offset untuk sumber Kafka saat sumber Kafka mulai membaca data. Inisialisasi offset adalah objek yang mengimplementasikan antarmuka OffsetsInitializer. Kelas KafkaSource menyediakan inisialisasi offset bawaan berikut.

    Inisialisasi offset	Pengaturan Kode
    Membaca data dari offset paling awal.	KafkaSource.builder().setStartingOffsets(OffsetsInitializer.earliest())
    Membaca data dari offset terbaru.	KafkaSource.builder().setStartingOffsets(OffsetsInitializer.latest())
    Memulai mengonsumsi data dengan timestamp lebih besar dari atau sama dengan waktu yang ditentukan (dalam milidetik).	KafkaSource.builder().setStartingOffsets(OffsetsInitializer.timestamp(1592323200000L))
    Mengonsumsi dari offset yang dikomit oleh kelompok konsumen. Jika tidak ada offset tersebut, gunakan offset paling awal.	KafkaSource.builder().setStartingOffsets(OffsetsInitializer.committedOffsets(OffsetResetStrategy.EARLIEST))
    Membaca data dari offset yang dikomit setiap partisi dan tidak ada strategi reset yang ditentukan.	KafkaSource.builder().setStartingOffsets(OffsetsInitializer.committedOffsets())

    Catatan

    • Jika inisialisasi offset bawaan tidak memenuhi kebutuhan bisnis Anda, Anda dapat membuat inisialisasi offset kustom.

    • Jika Anda tidak menentukan inisialisasi offset, inisialisasi offset OffsetsInitializer.earliest() digunakan secara default.

  • Mode eksekusi streaming dan mode eksekusi batch

    Sumber Kafka dapat beroperasi dalam mode streaming atau mode batch. Secara default, sumber Kafka beroperasi dalam mode streaming. Dalam mode ini, deployment terus berjalan hingga deployment gagal atau dibatalkan. Jika Anda ingin sumber Kafka beroperasi dalam mode batch, Anda dapat memanggil metode setBounded(OffsetsInitializer) untuk menentukan offset berhenti. Saat semua partisi mencapai offset berhentinya, sumber Kafka keluar.

    Catatan

    Dalam kebanyakan kasus, sumber Kafka yang beroperasi dalam mode streaming tidak memiliki offset berhenti. Jika Anda ingin men-debug sumber Kafka yang beroperasi dalam mode streaming, Anda dapat memanggil metode setUnbounded(OffsetsInitializer) untuk menentukan offset berhenti. Metode yang Anda gunakan untuk menentukan offset berhenti bervariasi tergantung pada apakah Anda menggunakan mode streaming atau mode batch.

  • Penemuan partisi dinamis

    Jika Anda ingin deployment yang sedang berjalan memproses data dari topik baru dan partisi baru yang cocok dengan pola langganan Anda tanpa me-restart deployment, Anda dapat mengaktifkan fitur penemuan partisi dinamis pada sumber Kafka. Dalam konektor DataStream, fitur ini dinonaktifkan secara default dan harus diaktifkan secara manual:

    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 segera memperbarui informasi partisi, 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 yang dilampirkan pada catatan sebagai waktu event untuk catatan tersebut. Anda dapat mendefinisikan strategi watermark berdasarkan waktu event setiap catatan dan mengirimkan watermark ke layanan hilir.

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

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

    Catatan

    Jika beberapa subtugas sumber tetap menganggur untuk jangka waktu yang lama, seperti partisi Kafka yang tidak menerima pesan baru atau konkurensi sumber melebihi jumlah partisi Kafka, pembuatan watermark mungkin gagal. Dalam kasus ini, komputasi window tidak dapat dipicu, dan pemrosesan data akan berhenti.

    Solusinya adalah sebagai berikut:

    • Konfigurasi mekanisme timeout watermark: Aktifkan parameter table.exec.source.idle-timeout untuk memaksa sistem menghasilkan watermark setelah periode timeout yang ditentukan, memastikan progres epoch komputasi window.

    • Optimalkan sumber data: Atur konkurensi sumber agar sama dengan atau kurang dari jumlah partisi Kafka.

  • Consumer Offset Commit

    Saat checkpoint dihasilkan, sumber Kafka mengirimkan offset konsumen Kafka setiap partisi ke broker Kafka. Hal ini memastikan bahwa offset konsumen Kafka yang dicatat di broker Kafka konsisten dengan status checkpoint. Konsumen Kafka dapat secara otomatis mengirimkan offset pada setiap partisi ke broker Kafka secara berkala. Anda dapat mengonfigurasi fitur pengiriman offset otomatis menggunakan opsi enable.auto.commit dan auto.commit.interval.ms. Jika Anda menonaktifkan fitur checkpointing, sumber Kafka mengandalkan konsumen Kafka untuk mengirimkan offset ke broker Kafka.

    Catatan

    Sumber Kafka tidak menggunakan offset yang dikirimkan yang dicatat di broker Kafka untuk toleransi kesalahan. Saat Anda mengirimkan offset, broker Kafka dapat memantau progres konsumsi catatan pada setiap partisi.

  • Properti tambahan

    Anda dapat memanggil metode setProperties(Properties) dan setProperty(String, String) untuk mengonfigurasi properti tambahan untuk sumber Kafka dan konsumen Kafka. Tabel berikut menjelaskan properti sumber Kafka.

    Item konfigurasi	Deskripsi
    client.id.prefix	Menentukan awalan untuk ID klien konsumen Kafka.
    partition.discovery.interval.ms	Menentukan interval waktu saat sumber Kafka memeriksa partisi baru. Catatan Properti partition.discovery.interval.ms ditimpa menjadi -1 dalam mode batch.
    register.consumer.metrics	Menentukan apakah akan mendaftarkan metrik untuk konsumen Kafka di Realtime Compute for Apache Flink.
    Konfigurasi Konsumen Kafka lainnya	Untuk informasi selengkapnya tentang properti konsumen Kafka, lihat Apache Kafka. Penting Konektor DataStream Kafka menimpa nilai properti berikut: key.deserializer: Nilai properti ini diatur ke ByteArrayDeserializer. value.deserializer: Nilai properti ini diatur ke ByteArrayDeserializer. auto.offset.reset.strategy: Nilai properti ini diatur ke OffsetsInitializer#getAutoOffsetResetStrategy().

    Kode contoh berikut menunjukkan cara konsumen Kafka terhubung ke kluster Kafka menggunakan konfigurasi JAAS dan mekanisme otentikasi SASL/PLAIN.

    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 Realtime Compute for Apache Flink untuk pemantauan dan diagnosis.

    • Cakupan metrik

      Semua metrik sumber Kafka didaftarkan di bawah grup metrik KafkaSourceReader. KafkaSourceReader adalah subgrup dari grup metrik operator. Metrik untuk partisi tertentu didaftarkan di grup metrik KafkaSourceReader.topic.<nama_topik>.partition.<id_partisi>.

      Misalnya, topik bernama my-topic dan partisi topik bernama 1. Offset konsumen partisi dilaporkan oleh metrik <some_parent_groups>.operator.KafkaSourceReader.topic.my-topic.partition.1.currentOffset. Jumlah pengiriman offset konsumen yang berhasil diukur oleh metrik <some_parent_groups>.operator.KafkaSourceReader.commitsSucceeded.

    • Metrik

      Metrik	Deskripsi	Cakupan
      currentOffset	Offset Konsumen Saat Ini	TopicPartition
      committedOffset	Offset komit saat ini	TopicPartition
      commitsSucceeded	Jumlah pengiriman yang berhasil	KafkaSourceReader
      commitsFailed	Jumlah Pengiriman Gagal	KafkaSourceReader

    • Metrik Konsumen Kafka

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

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

• Buat sink Kafka

  Sink Kafka dapat menulis data dari beberapa 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 tujuan
                                          .setKafkaValueSerializer(StringSerializer.class) // skema serialisasi
                                          .build())
                          .setDeliveryGuarantee(DeliveryGuarantee.EXACTLY_ONCE) // toleransi kesalahan
                          .build();
  
  stream.sinkTo(kafkaSink);

  Anda harus mengonfigurasi parameter berikut.

  Parameter	Deskripsi
  Topik	Nama topik tempat data ditulis.
  Serialisasi data	Saat Anda membuat sink Kafka, Anda harus menyediakan KafkaRecordSerializationSchema untuk mengonversi data input menjadi objek ProducerRecord Kafka. Flink menyediakan builder skema yang menawarkan komponen umum, seperti serialisasi kunci dan nilai pesan, pemilihan topik, dan partisi pesan. Anda juga dapat mengimplementasikan antarmuka yang sesuai untuk kontrol lanjutan. Sink Kafka memanggil metode ProducerRecord<byte[], byte[]> serialize(T element, KafkaSinkContext context, Long timestamp) untuk setiap catatan masuk untuk menghasilkan objek ProducerRecord yang merepresentasikan catatan yang telah diserialisasi. Kemudian, sink Kafka menulis objek ProducerRecord ke topik yang diperlukan. Anda dapat mengontrol cara setiap catatan ditulis ke Kafka secara detail. Menggunakan ProducerRecord, Anda dapat melakukan tindakan berikut: Menyetel nama topik tujuan. Menentukan kunci pesan. Menentukan partisi tujuan.
  Properti klien Kafka	Properti bootstrap.servers wajib diisi. Tentukan daftar alamat broker Kafka yang dipisahkan koma.
  Semantik toleransi kesalahan	Setelah Anda mengaktifkan fitur checkpointing, sink Kafka dapat memastikan pengiriman tepat-sekali. Anda juga dapat mengonfigurasi parameter DeliveryGuarantee untuk menentukan semantik toleransi kesalahan yang berbeda. Detail tentang parameter DeliveryGuarantee adalah sebagai berikut: DeliveryGuarantee.NONE: Tidak ada jaminan pengiriman yang disediakan oleh Flink. Data mungkin hilang atau diduplikasi. DeliveryGuarantee.AT_LEAST_ONCE: Sink Kafka memastikan data tidak hilang. Namun, data mungkin diduplikasi. DeliveryGuarantee.EXACTLY_ONCE: Sink Kafka memastikan data tidak hilang atau diduplikasi. Mekanisme transaksi Kafka digunakan untuk memastikan pengiriman tepat-sekali. Catatan Untuk informasi selengkapnya tentang semantik tepat-sekali, lihat Catatan penggunaan Semantic.EXACTLY_ONCE.

Toleransi dan pengumpulan data kotor

Dalam beberapa kasus, sumber data Kafka Anda mungkin berisi data rusak (data kotor). Untuk mencegah restart pekerjaan yang sering karena data kotor tersebut, Anda dapat mengonfigurasi pekerjaan untuk mengabaikan exception tersebut. Konfigurasi contoh:

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

Konfigurasi ini mentoleransi hingga 1000 catatan data kotor, memungkinkan pekerjaan Anda berjalan normal saat terdapat sedikit data kotor. Saat jumlah catatan data 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
  # Enable dirty data tolerance.
  ingestion.ignore-errors: true
  # Tolerate all dirty data.
  ingestion.error-tolerance.max-count: -1

Kebijakan toleransi data kotor mencegah pekerjaan gagal secara sering karena data abnormal. Anda mungkin juga ingin mempelajari lebih lanjut tentang data kotor untuk menyesuaikan perilaku produsen Kafka. Untuk proses yang dijelaskan dalam Pengumpulan Data Kotor, Anda dapat melihat data kotor pekerjaan Anda di log TaskManager. Kode berikut menunjukkan contoh konfigurasi:

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.
  ingestion.error-tolerance.max-count: -1

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

Strategi pemetaan nama tabel dan topik

Saat menggunakan Kafka sebagai sink ingesti data, format pesan (debezium-json atau canal-json) sering kali mencakup informasi nama tabel. Konsumen biasanya menggunakan nama tabel ini—bukan nama topik—sebagai nama tabel sebenarnya. Oleh karena itu, konfigurasikan strategi pemetaan nama tabel dan topik dengan hati-hati.

Asumsikan Anda perlu menyinkronkan dua tabel—mydb.mytable1 dan mydb.mytable2—dari MySQL. Strategi pemetaan yang mungkin termasuk yang berikut:

1. Jangan mengonfigurasi strategi pemetaan apa pun

Tanpa strategi pemetaan apa pun, setiap tabel ditulis ke topik yang dinamai berdasarkan database dan tabel (misalnya, mydb.mytable1). Dengan demikian, data mydb.mytable1 masuk ke topik mydb.mytable1, dan data mydb.mytable2 masuk ke topik mydb.mytable2. Konfigurasi contoh:

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 rute untuk pemetaan (tidak disarankan)

Dalam banyak skenario, pengguna tidak ingin topik dinamai berdasarkan database dan tabel. Sebaliknya, mereka mengonfigurasi aturan rute untuk memetakan data ke topik tertentu. Konfigurasi contoh:

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, memodifikasi nama topik melalui aturan rute juga memodifikasi nama tabel dalam pesan Kafka (dalam format debezium-json atau canal-json). Akibatnya, semua pesan dalam topik ini memiliki nama tabel mytable1. Sistem lain yang mengonsumsi topik ini mungkin berperilaku tidak terduga.

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

Untuk mempertahankan informasi nama tabel asli sambil memetakan ke topik kustom, gunakan parameter sink.tableId-to-topic.mapping. Konfigurasi contoh:

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 konfigurasi ini, semua data dari `mydb.mytable1` dan `mydb.mytable2` ditulis ke topik `mytable1`, tetapi nama tabel asli (`mydb.mytable1` atau `mydb.mytable2`) dipertahankan dalam format pesan Kafka (`debezium-json` atau `canal-json`). Hal ini memungkinkan sistem lain yang mengonsumsi pesan dari topik ini untuk mengambil informasi nama tabel sumber dengan benar.

Pertimbangan 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): Memungkinkan Anda 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 memulihkan dari checkpoint, Flink hanya bergantung pada transaksi yang dikomit sebelum checkpoint dimulai. Jika waktu antara crash pekerjaan dan restart melebihi timeout transaksi Kafka, Kafka secara otomatis membatalkan transaksi, menyebabkan kehilangan data.

  • Default broker Kafka transaction.max.timeout.ms = 15 menit.

  • Sink Kafka Flink mengatur transaction.timeout.ms ke 1 jam secara default.

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

• Kolam produsen Kafka dan checkpoint konkuren

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

  Sesuaikan ukuran kolam produsen sesuai dengan jumlah maksimum checkpoint konkuren Anda.

• Scale-in restrictions

  Jika pekerjaan gagal sebelum checkpoint pertama, informasi kolam produsen tidak dipertahankan setelah restart. Oleh karena itu, sebelum checkpoint pertama selesai, jangan menurunkan konkurensi pekerjaan. Jika Anda harus menurunkan skala, konkurensi tidak boleh kurang dari FlinkKafkaProducer.SAFE_SCALE_DOWN_FACTOR.

• Pemblokiran pembacaan transaksional

  Dalam mode read_committed, transaksi yang belum selesai (tidak dikomit maupun dibatalkan) memblokir pembacaan dari seluruh topik.

  Misalnya:

  • Transaksi 1 menulis data.

  • Transaksi 2 menulis dan melakukan commit data.

  • Hingga Transaksi 1 selesai, data Transaksi 2 tidak terlihat oleh konsumen.

  Oleh karena itu:

  • Selama operasi normal, visibilitas data tertunda sekitar interval rata-rata checkpoint.

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

FAQ

• Bagaimana cara mendapatkan data JSON di Kafka menggunakan Realtime Compute for Apache Flink?

• Mengapa tidak ada output data setelah data tabel sumber Kafka diproses menggunakan window berbasis waktu event?

• Bagaimana cara menghubungkan ke kluster Kafka yang telah dikonfigurasi informasi keamanannya?

• Bagaimana cara menyelesaikan konflik penamaan bidang?

• Manajemen dan operasi penyimpanan