DataWorks：Sumber data Kafka

Modul data record 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 apendiks.

value

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

offset

Long

timestamp

Long

headers

String

partition

Long

Jenis tugas sinkronisasi

Format value 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 dianggap sebagai string yang dikodekan UTF-8 dan dikonversi menjadi string.

text

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 dianggap 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.

text

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

Anda perlu mengonfigurasi org.apache.kafka.common.security.plain.PlainLoginModule.

username

Username. Konfigurasikan item ini sesuai kebutuhan.

password

Password. 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.

Tidak

valueType

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

Tidak

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 secara berkala. Oleh karena itu, jika tugas gagal dan dijalankan ulang, data mungkin diduplikasi atau hilang. Jika parameter skipExceedRecord diatur ke true, tugas mungkin membuang beberapa record terakhir yang telah 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 komit 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 hanya membaca data dari rentang offset satu partisi tunggal.

Tidak. Tidak ada nilai default.

kafkaConfig

Saat membuat klien KafkaConsumer untuk konsumsi data, Anda dapat menentukan parameter ekstensi 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 percobaan.

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 dalam topik telah dibaca, atau karena masalah ketersediaan jaringan atau 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 dalam 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 ada cukup data.

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 yang dapat dilewati konsumen 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 record dengan offset tersebut telah kedaluwarsa dan dihapus). Nilai default adalah none, yang berarti offset tidak direset secara otomatis. Anda dapat mengubahnya menjadi earliest, yang berarti konsumen membaca record 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

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

security.protocol

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

sasl.mechanism

Metode autentikasi SASL. Jika Anda menggunakan Alibaba Cloud Kafka, 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 secara default untuk membentuk nilai. Pemisah ditentukan oleh fieldDelimiter.

Tidak

writeMode

Saat valueIndex tidak dikonfigurasi, parameter ini menentukan format penggabungan semua kolom record sumber untuk membentuk nilai record 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 record sumber memiliki tiga kolom dengan nilai a, b, dan c, dan writeMode diatur ke text serta fieldDelimiter diatur ke #, nilai record Kafka yang ditulis adalah string a#b#c. Jika writeMode diatur ke JSON dan column diatur ke [{"name":"col1"},{"name":"col2"},{"name":"col3"}], nilai record 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 record sumber. Contoh, "column": [{"name":"id","type":"JSON_NUMBER"}, {"name":"name","type":"JSON_STRING"}, {"name":"age","type":"JSON_NUMBER"}].

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

  Jika record 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 record Kafka yang ditulis adalah string {"col1":"a","col2":"b"}.

• Jika jumlah kolom dalam record sumber lebih kecil 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 record 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 record 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 Apendiks: 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 ordinal kolom dalam record sumber yang digunakan sebagai kunci untuk record Kafka.

Nomor ordinal kolom dimulai dari 0. Misalnya, [0,1,2] akan menggabungkan nilai semua nomor kolom yang dikonfigurasi dengan koma untuk membentuk kunci record Kafka. Jika tidak ditentukan, kunci record 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 record sumber digabungkan menggunakan pemisah kolom yang ditentukan oleh parameter ini untuk membentuk nilai record 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 record Kafka. Jika tidak dikonfigurasi, tidak ada penggantian yang dilakukan.

Tidak

acks

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

• 0: Tidak ada acknowledgment untuk penulisan yang berhasil.

• 1: Acknowledgment untuk penulisan yang berhasil ke replica primer.

• all: Acknowledgment untuk penulisan yang berhasil ke semua replica.

Tidak