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 field sumber

Metode pemrosesan untuk operasi penulisan

Sinkronisasi offline

Node sinkronisasi offline di DataStudio

json

String

String yang dikodekan UTF-8

Boolean

Dikonversi ke 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.

text

String

String yang dikodekan UTF-8

Boolean

Dikonversi ke 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 ke bilangan bulat JSON 13 digit yang merepresentasikan timestamp dalam milidetik.

• Untuk nilai waktu dengan presisi mikrodetik atau nanodetik: Dikonversi ke 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 ke string yang dikodekan UTF-8.

text

String

String yang dikodekan UTF-8

Boolean

Dikonversi ke 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 ke 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 ke 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 ke string yang dikodekan UTF-8.

Item konfigurasi

Deskripsi

Modul logon

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

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 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 mengambil properti objek JSON, seperti ["event_id"].

  • Jika data Anda dalam format JSON, Anda dapat mengambil 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

Waktu mulai untuk konsumsi data. Parameter ini menentukan batas kiri rentang waktu, yang inklusif. Ini adalah string waktu dalam format yyyymmddhhmmss. Anda dapat menggunakan parameter ini dengan scheduling parameters. Untuk informasi lebih lanjut, lihat Format yang didukung untuk parameter penjadwalan.

Anda harus menentukan salah satu parameter ini atau beginOffset.

endDateTime

Waktu akhir untuk konsumsi data. Parameter ini menentukan batas kanan rentang waktu, yang eksklusif. Ini adalah string waktu dalam format yyyymmddhhmmss. Anda dapat menggunakan parameter ini 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: Membaca data mulai dari offset yang disimpan untuk ID grup yang ditentukan oleh group.id dalam parameter 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, duplikasi atau kehilangan data mungkin terjadi. Jika parameter skipExceedRecord diatur ke true, tugas mungkin membuang beberapa catatan 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 dapat mengambil data yang melebihi endOffset atau endDateTime yang ditentukan. Parameter ini menentukan apakah data berlebih ini ditulis ke tujuan. Karena tugas 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 versi selanjutnya: 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.

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 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 di 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 saat stopWhenPollEmpty adalah true. Nilai yang valid adalah true dan false.

• Jika parameter ini diatur ke true dan konsumen tidak menerima data dari permintaan poll, sistem memeriksa apakah offset terbaru di partisi topik telah tercapai. Jika offset terbaru telah tercapai untuk semua partisi, tugas berhenti segera. Jika tidak, tugas terus mencoba menarik data dari topik.

• Jika parameter ini diatur ke false dan konsumen tidak menerima data dari permintaan poll, sistem tidak melakukan pemeriksaan dan tugas berhenti segera.

Tidak. Nilai default adalah false.

Parameter

Deskripsi

fetch.min.bytes

Jumlah minimum data, dalam byte, yang diambil konsumen dari broker dalam satu permintaan. Broker menunggu hingga jumlah data ini tersedia sebelum merespons konsumen.

fetch.max.wait.ms

Waktu maksimum, dalam milidetik, yang ditunggu broker agar data tersedia sebelum merespons permintaan fetch. Nilai default adalah 500. Broker merespons ketika salah satu kondisi fetch.min.bytes atau fetch.max.wait.ms terpenuhi.

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 catatan dengan offset tersebut telah kedaluwarsa dan dihapus). Nilai default adalah none, yang berarti offset tidak diatur ulang secara otomatis. Anda dapat mengubahnya ke earliest, yang berarti konsumen membaca catatan partition dari offset paling awal.

max.poll.records

Jumlah pesan yang dapat dikembalikan oleh satu panggilan ke 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 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 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 field yang ditentukan oleh parameter column.

Misalnya, jika catatan sumber memiliki tiga kolom dengan nilai a, b, dan c, dan writeMode diatur ke text dan 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

Field 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 menentukan nama bidang dalam struktur JSON untuk nilai kolom dari catatan sumber. Contohnya: "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 field 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 record Kafka yang ditulis adalah string {"col1":"a","col2":"b"}.

• Jika jumlah kolom dalam catatan sumber lebih sedikit daripada jumlah field yang ditentukan dalam column, field tambahan 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 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 field JSON tidak dikonfigurasi, tipe field default adalah JSON_STRING.

• Untuk nilai yang valid dari tipe field JSON, lihat Apendiks: Tipe field 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 di 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 catatan 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 catatan 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 Anda tidak menentukan parameter ini, 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 replika primer.

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

Tidak