DataWorks：Sumber data Kafka

Modul data catatan Kafka

Tipe data yang diproses

key

Bergantung pada item konfigurasi keyType dalam tugas sinkronisasi data. Untuk informasi lebih lanjut tentang parameter keyType, lihat deskripsi parameter lengkap di lampiran.

value

Bergantung pada item konfigurasi valueType dalam tugas sinkronisasi data. Untuk informasi lebih lanjut tentang parameter valueType, lihat deskripsi parameter lengkap di lampiran.

offset

Long

timestamp

Long

headers

String

partition

Long

Jenis tugas sinkronisasi

Format nilai yang ditulis ke Kafka

Tipe bidang sumber

Metode pemrosesan untuk operasi tulis

Sinkronisasi offline

Node sinkronisasi offline di DataStudio

json

String

String yang dikodekan UTF-8

Boolean

Dikonversi menjadi string "true" atau "false" yang dikodekan UTF-8

Waktu/Tanggal

String yang dikodekan UTF-8 dalam format yyyy-MM-dd HH:mm:ss

Numerik

String numerik yang dikodekan UTF-8

Aliran byte

Aliran byte diperlakukan sebagai string yang dikodekan UTF-8 dan dikonversi menjadi string.

teks

String

String yang dikodekan UTF-8

Boolean

Dikonversi menjadi string "true" atau "false" yang dikodekan UTF-8

Waktu/Tanggal

String yang dikodekan UTF-8 dalam format yyyy-MM-dd HH:mm:ss

Numerik

String numerik yang dikodekan UTF-8

Aliran byte

Aliran byte diperlakukan sebagai string yang dikodekan UTF-8 dan dikonversi menjadi string.

Sinkronisasi real-time: ETL real-time ke Kafka

Node sinkronisasi real-time di DataStudio

json

String

String yang dikodekan UTF-8

Boolean

Tipe Boolean JSON

Waktu/Tanggal

• Untuk nilai waktu dengan presisi kurang dari milidetik: Dikonversi menjadi bilangan bulat JSON 13 digit yang merepresentasikan timestamp dalam milidetik.

• Untuk nilai waktu dengan presisi mikrodetik atau nanodetik: Dikonversi menjadi bilangan titik mengambang JSON yang mencakup bilangan bulat 13 digit untuk timestamp milidetik dan desimal 6 digit untuk timestamp nanodetik.

Numerik

Tipe numerik JSON

Aliran byte

Aliran byte dikodekan Base64 lalu dikonversi menjadi string yang dikodekan UTF-8.

teks

String

String yang dikodekan UTF-8

Boolean

Dikonversi menjadi string "true" atau "false" yang dikodekan UTF-8

Waktu/Tanggal

String yang dikodekan UTF-8 dalam format yyyy-MM-dd HH:mm:ss

Numerik

String numerik yang dikodekan UTF-8

Aliran byte

Aliran byte dikodekan Base64 lalu dikonversi menjadi string yang dikodekan UTF-8.

Sinkronisasi real-time: Sinkronisasi real-time seluruh database ke Kafka

Hanya sinkronisasi data inkremental

Format JSON bawaan

String

String yang dikodekan UTF-8

Boolean

Tipe Boolean JSON

Waktu/Tanggal

Timestamp milidetik 13 digit

Numerik

Nilai numerik JSON

Aliran byte

Aliran byte dikodekan Base64 lalu dikonversi menjadi string yang dikodekan UTF-8.

Solusi sinkronisasi: Sinkronisasi real-time sekali klik ke Kafka

Sinkronisasi offline penuh + sinkronisasi real-time inkremental

Format JSON bawaan

String

String yang dikodekan UTF-8

Boolean

Tipe Boolean JSON

Waktu/Tanggal

Timestamp milidetik 13 digit

Numerik

Nilai numerik JSON

Aliran byte

Aliran byte dikodekan Base64 lalu dikonversi menjadi string yang dikodekan UTF-8.

Item konfigurasi

Deskripsi

Modul logon

Harus diatur ke org.apache.kafka.common.security.plain.PlainLoginModul

username

Nama pengguna. Konfigurasikan item ini sesuai kebutuhan.

password

Kata sandi. Konfigurasikan item ini sesuai kebutuhan.

Parameter

Deskripsi

Wajib

datasource

Nama sumber data. Editor kode mendukung penambahan sumber data. Nilai parameter ini harus sama dengan nama sumber data yang ditambahkan.

Ya

server

Alamat server broker Kafka dalam format ip:port.

Anda hanya dapat mengonfigurasi satu server, tetapi Anda harus memastikan bahwa DataWorks dapat terhubung ke alamat IP semua broker di kluster Kafka.

Ya

topic

Topik Kafka. Topik adalah agregasi feed pesan yang diproses Kafka.

Ya

column

Data Kafka yang akan dibaca. Kolom konstan, kolom data, dan kolom atribut didukung.

• Kolom konstan: Kolom yang diapit tanda kutip tunggal, seperti ["'abc'", "'123'"].

• Kolom data

  • Jika data Anda dalam format JSON, Anda dapat mendapatkan properti objek JSON, seperti ["event_id"].

  • Jika data Anda dalam format JSON, Anda dapat mendapatkan sub-properti bersarang objek JSON, seperti ["tag.desc"].

• Kolom atribut

  • __key__: kunci pesan.

  • __value__: konten lengkap pesan.

  • __partition__: partisi tempat pesan saat ini berada.

  • __headers__: header pesan saat ini.

  • __offset__: offset pesan saat ini.

  • __timestamp__: timestamp pesan saat ini.

  Kode berikut memberikan contoh lengkap.

  "column": [
      "__key__",
      "__value__",
      "__partition__",
      "__offset__",
      "__timestamp__",
      "'123'",
      "event_id",
      "tag.desc"
      ]

Ya

keyType

Tipe kunci Kafka. Nilai yang valid: BYTEARRAY, DOUBLE, FLOAT, INTEGER, LONG, dan SHORT.

Ya

valueType

Tipe nilai Kafka. Nilai yang valid: BYTEARRAY, DOUBLE, FLOAT, INTEGER, LONG, dan SHORT.

Ya

beginDateTime

Offset awal untuk konsumsi data. Ini adalah batas kiri rentang waktu (inklusif). Ini adalah string waktu dalam format yyyymmddhhmmss. Anda dapat menggunakannya dengan Scheduling Parameters. Untuk informasi lebih lanjut, lihat Format yang didukung untuk parameter penjadwalan.

Anda harus menentukan salah satu parameter ini atau beginOffset.

endDateTime

Offset akhir untuk konsumsi data. Ini adalah batas kanan rentang waktu (eksklusif). Ini adalah string waktu dalam format yyyymmddhhmmss. Anda dapat menggunakannya dengan Scheduling Parameters. Untuk informasi lebih lanjut, lihat Format yang didukung untuk parameter penjadwalan.

Anda harus menentukan salah satu parameter ini atau endOffset.

beginOffset

Offset awal untuk konsumsi data. Anda dapat mengonfigurasinya dalam bentuk berikut:

• Bilangan, seperti 15553274, yang menunjukkan offset awal untuk konsumsi.

• seekToBeginning: menunjukkan bahwa data dikonsumsi dari offset paling awal.

• seekToLast: menunjukkan bahwa data dibaca dari offset yang disimpan untuk ID grup yang ditentukan oleh group.id dalam konfigurasi kafkaConfig. Perhatikan bahwa offset grup secara otomatis dikomit ke server Kafka oleh klien pada interval reguler. Oleh karena itu, jika tugas gagal dan dijalankan ulang, data mungkin diduplikasi atau hilang. Jika parameter skipExceedRecord diatur ke true, tugas mungkin membuang beberapa catatan terakhir yang dibaca. Offset grup untuk data yang dibuang ini telah dikomit ke server, sehingga data ini tidak dapat dibaca dalam eksekusi tugas berikutnya.

• seekToEnd: menunjukkan bahwa data dikonsumsi dari offset terbaru. Ini akan membaca data kosong.

Anda harus menentukan salah satu parameter ini atau beginDateTime.

endOffset

Offset akhir untuk konsumsi data. Ini digunakan untuk mengontrol kapan tugas konsumsi data berhenti.

Anda harus menentukan salah satu parameter ini atau endDateTime.

skipExceedRecord

Kafka menggunakan public ConsumerRecords<K, V> poll(final Duration timeout) untuk mengonsumsi data. Satu panggilan poll tunggal mungkin mengambil data yang berada di luar rentang endOffset atau endDateTime. skipExceedRecord mengontrol apakah data berlebih ditulis ke tujuan. Karena konsumsi data menggunakan komitmen offset otomatis, kami merekomendasikan hal berikut:

• Untuk versi Kafka sebelum 0.10.2: Atur skipExceedRecord ke false.

• Untuk Kafka 0.10.2 dan yang lebih baru: Atur skipExceedRecord ke true.

Tidak. Nilai default adalah false.

partition

Topik Kafka memiliki beberapa partisi (partition). Secara default, tugas sinkronisasi data membaca data dari rentang offset yang mencakup semua partisi dalam topik. Anda juga dapat menentukan partition untuk membaca data hanya dari rentang offset satu partisi saja.

Tidak. Tidak ada nilai default.

kafkaConfig

Saat membuat klien KafkaConsumer untuk konsumsi data, Anda dapat menentukan parameter tambahan seperti bootstrap.servers, auto.commit.interval.ms, dan session.timeout.ms. Anda dapat menggunakan kafkaConfig untuk mengontrol perilaku konsumsi KafkaConsumer.

Tidak

encoding

Saat keyType atau valueType diatur ke STRING, encoding yang ditentukan oleh parameter ini digunakan untuk mengurai string.

Tidak. Nilai default adalah UTF-8.

waitTIme

Waktu maksimum, dalam detik, yang ditunggu objek konsumen untuk menarik data dari Kafka dalam satu upaya.

Tidak. Nilai default adalah 60.

stopWhenPollEmpty

Nilai yang valid adalah true dan false. Jika parameter ini diatur ke true dan konsumen menarik data kosong dari Kafka (biasanya karena semua data di topik telah dibaca, atau karena masalah jaringan atau ketersediaan kluster Kafka), tugas berhenti segera. Jika tidak, tugas mencoba lagi hingga data berhasil dibaca.

Tidak. Nilai default adalah true.

stopWhenReachEndOffset

Parameter ini hanya berlaku ketika stopWhenPollEmpty adalah true. Nilai yang valid adalah true dan false.

• Jika parameter ini diatur ke true dan konsumen menarik data kosong dari Kafka, sistem memeriksa apakah data terbaru di partisi topik Kafka telah dibaca. Jika data terbaru dari semua partisi telah dibaca, tugas berhenti segera. Jika tidak, sistem terus mencoba menarik data dari topik Kafka.

• Jika parameter ini diatur ke false dan konsumen menarik data kosong dari Kafka, sistem tidak melakukan pemeriksaan dan langsung menghentikan tugas.

Tidak. Nilai default adalah false.

Parameter

Deskripsi

fetch.min.bytes

Menentukan jumlah minimum byte pesan yang dapat diperoleh konsumen dari broker. Data dikembalikan ke konsumen hanya ketika tersedia data yang cukup.

fetch.max.wait.ms

Waktu maksimum menunggu broker mengembalikan data. Nilai default adalah 500 milidetik. Data dikembalikan berdasarkan kondisi mana yang terpenuhi lebih dulu: fetch.min.bytes atau fetch.max.wait.ms.

max.partition.fetch.bytes

Menentukan jumlah maksimum byte yang dapat dikembalikan broker ke konsumen dari setiap partition. Nilai default adalah 1 MB.

session.timeout.ms

Menentukan waktu maksimum konsumen dapat terputus dari server sebelum berhenti menerima layanan. Nilai default adalah 30 detik.

auto.offset.reset

Tindakan yang diambil konsumen saat membaca tanpa offset atau offset tidak valid (karena konsumen tidak aktif dalam waktu lama dan catatan dengan offset tersebut telah kedaluwarsa dan dihapus). Nilai default adalah none, yang berarti offset tidak diatur ulang secara otomatis. Anda dapat mengubahnya menjadi earliest, yang berarti konsumen membaca catatan partition dari offset paling awal.

max.poll.records

Jumlah pesan yang dapat dikembalikan oleh satu panggilan metode poll.

key.deserializer

Metode deserialisasi untuk kunci pesan, seperti org.apache.kafka.common.serialization.StringDeserializer.

value.deserializer

Metode deserialisasi untuk nilai data, seperti org.apache.kafka.common.serialization.StringDeserializer.

ssl.truststore.location

Path sertifikat root SSL.

ssl.truststore.password

Kata sandi untuk penyimpanan sertifikat root. Jika Anda menggunakan Kafka Alibaba Cloud, atur ini ke KafkaOnsClient.

security.protocol

Protokol akses. Saat ini, hanya protokol SASL_SSL yang didukung.

sasl.mechanism

Metode autentikasi SASL. Jika Anda menggunakan Kafka Alibaba Cloud, gunakan PLAIN.

java.security.auth.login.config

Path file autentikasi SASL.

Parameter

Deskripsi

Wajib

datasource

Nama sumber data. Editor kode mendukung penambahan sumber data. Nilai parameter ini harus sama dengan nama sumber data yang ditambahkan.

Ya

server

Alamat server Kafka dalam format ip:port.

Ya

topic

Topik Kafka. Ini adalah kategori untuk feed pesan berbeda yang diproses Kafka.

Setiap pesan yang dipublikasikan ke kluster Kafka memiliki kategori, yang disebut topik. Topik adalah kumpulan sekelompok pesan.

Ya

valueIndex

Kolom di writer Kafka yang digunakan sebagai nilai. Jika tidak ditentukan, semua kolom digabungkan untuk membentuk nilai secara default. Pemisah ditentukan oleh fieldDelimiter.

Tidak

writeMode

Saat valueIndex tidak dikonfigurasi, parameter ini menentukan format penggabungan semua kolom catatan sumber untuk membentuk nilai catatan Kafka. Nilai yang valid adalah text dan JSON. Nilai default adalah text.

• Jika diatur ke text, semua kolom digabungkan menggunakan pemisah yang ditentukan oleh fieldDelimiter.

• Jika diatur ke JSON, semua kolom digabungkan menjadi string JSON berdasarkan nama bidang yang ditentukan oleh parameter column.

Misalnya, jika catatan sumber memiliki tiga kolom dengan nilai a, b, dan c, dan writeMode diatur ke text serta fieldDelimiter diatur ke #, nilai catatan Kafka yang ditulis adalah string a#b#c. Jika writeMode diatur ke JSON dan column diatur ke [{"name":"col1"},{"name":"col2"},{"name":"col3"}], nilai catatan Kafka yang ditulis adalah string {"col1":"a","col2":"b","col3":"c"}.

Jika valueIndex dikonfigurasi, parameter ini tidak berlaku.

Tidak

column

Bidang di tabel tujuan tempat data akan ditulis, dipisahkan koma. Contoh: "column": ["id", "name", "age"].

Saat valueIndex tidak dikonfigurasi dan writeMode diatur ke JSON, parameter ini mendefinisikan nama bidang dalam struktur JSON untuk nilai kolom catatan sumber. Contoh, "column": [{"name":"id","type":"JSON_NUMBER"}, {"name":"name","type":"JSON_STRING"}, {"name":"age","type":"JSON_NUMBER"}].

• Jika jumlah kolom dalam catatan sumber lebih besar daripada jumlah nama bidang yang dikonfigurasi dalam column, data dipotong saat penulisan. Contoh:

  Jika catatan sumber memiliki tiga kolom dengan nilai a, b, dan c, dan column dikonfigurasi sebagai [{"name":"col1","type":"JSON_STRING"},{"name":"col2","type":"JSON_STRING"}], nilai catatan Kafka yang ditulis adalah string {"col1":"a","col2":"b"}.

• Jika jumlah kolom dalam catatan sumber lebih sedikit daripada jumlah nama bidang yang dikonfigurasi dalam column, nama bidang tambahan dalam konfigurasi column diisi dengan null atau string yang ditentukan oleh nullValueFormat. Contoh:

  Jika catatan sumber memiliki dua kolom dengan nilai a dan b, dan column dikonfigurasi sebagai [{"name":"col1","type":"JSON_STRING"},{"name":"col2","type":"JSON_STRING"},{"name":"col3","type":"JSON_STRING"}], nilai catatan Kafka yang ditulis adalah string {"col1":"a","col2":"b","col3":null}. Jika valueIndex dikonfigurasi, atau jika writeMode diatur ke text, parameter ini tidak berlaku.

• Jika tipe bidang JSON tidak dikonfigurasi, tipe bidang default adalah JSON_STRING.

• Untuk nilai yang valid dari tipe bidang JSON, lihat Lampiran: Tipe bidang JSON.

Jika valueIndex dikonfigurasi, atau jika writeMode diatur ke text, parameter ini tidak berlaku.

Wajib saat valueIndex tidak dikonfigurasi dan writeMode diatur ke JSON.

partition

Menentukan nomor partisi dalam topik Kafka tempat data ditulis. Ini harus berupa bilangan bulat yang lebih besar dari atau sama dengan 0.

Tidak

keyIndex

Kolom di writer Kafka yang digunakan sebagai kunci.

Nilai parameter keyIndex harus berupa bilangan bulat yang lebih besar dari atau sama dengan 0. Jika tidak, tugas akan gagal.

Tidak

keyIndexes

Array nomor urut kolom dalam catatan sumber yang digunakan sebagai kunci untuk catatan Kafka.

Nomor urut kolom dimulai dari 0. Misalnya, [0,1,2] akan menggabungkan nilai semua nomor kolom yang dikonfigurasi dengan koma untuk membentuk kunci catatan Kafka. Jika tidak ditentukan, kunci catatan Kafka adalah null, dan data ditulis ke partisi topik secara round-robin. Anda hanya dapat menentukan salah satu parameter ini atau keyIndex.

Tidak

fieldDelimiter

Saat writeMode diatur ke text dan valueIndex tidak dikonfigurasi, semua kolom catatan sumber digabungkan menggunakan pemisah kolom yang ditentukan oleh parameter ini untuk membentuk nilai catatan Kafka. Anda dapat mengonfigurasi satu karakter atau beberapa karakter sebagai pemisah. Anda dapat mengonfigurasi karakter Unicode dalam format \u0001. Karakter escape seperti \t dan \n didukung. Nilai default adalah \t.

Jika writeMode tidak diatur ke text atau jika valueIndex dikonfigurasi, parameter ini tidak berlaku.

Tidak

keyType

Tipe kunci Kafka. Nilai yang valid: BYTEARRAY, DOUBLE, FLOAT, INTEGER, LONG, dan SHORT.

Ya

valueType

Tipe nilai Kafka. Nilai yang valid: BYTEARRAY, DOUBLE, FLOAT, INTEGER, LONG, dan SHORT.

Ya

nullKeyFormat

Jika nilai kolom sumber yang ditentukan oleh keyIndex atau keyIndexes adalah null, nilai tersebut diganti dengan string yang ditentukan oleh parameter ini. Jika tidak dikonfigurasi, tidak ada penggantian yang dilakukan.

Tidak

nullValueFormat

Jika nilai kolom sumber adalah null, nilai tersebut diganti dengan string yang ditentukan oleh parameter ini saat merakit nilai catatan Kafka. Jika tidak dikonfigurasi, tidak ada penggantian yang dilakukan.

Tidak

acks

Konfigurasi acks saat menginisialisasi produsen Kafka. Ini menentukan metode pengakuan untuk penulisan yang berhasil. Secara default, parameter acks diatur ke all. Nilai yang valid untuk acks adalah:

• 0: Tidak ada pengakuan untuk penulisan yang berhasil.

• 1: Pengakuan untuk penulisan yang berhasil ke replika primer.

• all: Pengakuan untuk penulisan yang berhasil ke semua replika.

Tidak