DataWorks:Kafka veri kaynağı

Kafka kayıt veri modülü

İşlenen veri türü

key

Veri senkronizasyon görevindeki keyType yapılandırma öğesine bağlıdır. keyType parametresi hakkında daha fazla bilgi için ekte yer alan tam parametre açıklamasına bakın.

value

Veri senkronizasyon görevindeki valueType yapılandırma öğesine bağlıdır. valueType parametresi hakkında daha fazla bilgi için ekte yer alan tam parametre açıklamasına bakın.

offset

Long

timestamp

Long

headers

Dize

partition

Long

Senkronizasyon görevi türü

Kafka'ya yazılan value biçimi

Kaynak alan türü

Yazma işlemleri için işleme yöntemi

Çevrimdışı senkronizasyon

DataStudio'daki çevrimdışı senkronizasyon düğümü

json

Dize

UTF-8 kodlu dize

Boolean

"true" veya "false" UTF-8 kodlu dizesine dönüştürülür

Saat/Tarih

yyyy-MM-dd HH:mm:ss biçiminde UTF-8 kodlu dize

Sayısal

UTF-8 kodlu sayısal dize

Bayt akışı

Bayt akışı, UTF-8 kodlu bir dize olarak kabul edilir ve dizeye dönüştürülür.

text

Dize

UTF-8 kodlu dize

Boolean

"true" veya "false" UTF-8 kodlu dizesine dönüştürülür

Saat/Tarih

yyyy-MM-dd HH:mm:ss biçiminde UTF-8 kodlu dize

Sayısal

UTF-8 kodlu sayısal dize

Bayt akışı

Bayt akışı, UTF-8 kodlu bir dize olarak kabul edilir ve dizeye dönüştürülür.

Gerçek zamanlı senkronizasyon: Kafka'ya gerçek zamanlı ETL

DataStudio'daki gerçek zamanlı senkronizasyon düğümü

json

Dize

UTF-8 kodlu dize

Boolean

JSON Boolean türü

Saat/Tarih

• Milisaniyeden daha az hassasiyete sahip saat değerleri için: Milisaniye cinsinden zaman damgasını temsil eden 13 basamaklı bir JSON tamsayısına dönüştürülür.

• Mikrosaniye veya nanosaniye hassasiyetine sahip saat değerleri için: Milisaniye zaman damgası için 13 basamaklı bir tamsayı ve nanosaniye zaman damgası için 6 basamaklı bir ondalık sayı içeren bir JSON kayan noktalı sayısına dönüştürülür.

Sayısal

JSON sayısal türü

Bayt akışı

Bayt akışı Base64 ile kodlanır ve ardından UTF-8 kodlu bir dizeye dönüştürülür.

text

Dize

UTF-8 kodlu dize

Boolean

"true" veya "false" UTF-8 kodlu dizesine dönüştürülür

Saat/Tarih

yyyy-MM-dd HH:mm:ss biçiminde UTF-8 kodlu dize

Sayısal

UTF-8 kodlu sayısal dize

Bayt akışı

Bayt akışı Base64 ile kodlanır ve ardından UTF-8 kodlu bir dizeye dönüştürülür.

Gerçek zamanlı senkronizasyon: Tüm veritabanının Kafka'ya gerçek zamanlı senkronizasyonu

Yalnızca artımlı verilerin gerçek zamanlı senkronizasyonu

Yerleşik JSON biçimi

Dize

UTF-8 kodlu dize

Boolean

JSON Boolean türü

Saat/Tarih

13 basamaklı milisaniye zaman damgası

Sayısal

JSON sayısal değeri

Bayt akışı

Bayt akışı Base64 ile kodlanır ve ardından UTF-8 kodlu bir dizeye dönüştürülür.

Senkronizasyon çözümü: Kafka'ya tek tıkla gerçek zamanlı senkronizasyon

Tam çevrimdışı senkronizasyon + artımlı gerçek zamanlı senkronizasyon

Yerleşik JSON biçimi

Dize

UTF-8 kodlu dize

Boolean

JSON Boolean türü

Saat/Tarih

13 basamaklı milisaniye zaman damgası

Sayısal

JSON sayısal değeri

Bayt akışı

Bayt akışı Base64 ile kodlanır ve ardından UTF-8 kodlu bir dizeye dönüştürülür.

Yapılandırma öğesi

Açıklama

Oturum açma modülü

org.apache.kafka.common.security.plain.PlainLoginModul olarak ayarlanmalıdır

username

Kullanıcı adı. Bu öğeyi ihtiyaca göre yapılandırın.

password

Parola. Bu öğeyi ihtiyaca göre yapılandırın.

Parametre

Açıklama

Gerekli

datasource

Veri kaynağının adı. Kod düzenleyici, veri kaynakları eklemeyi destekler. Bu parametrenin değeri, eklenen veri kaynağının adıyla aynı olmalıdır.

Evet

server

Kafka broker sunucusunun adresi, ip:port biçimindedir.

Yalnızca bir server yapılandırabilirsiniz ancak DataWorks'ün Kafka kümesindeki tüm brokerların IP adreslerine bağlanabildiğinden emin olmalısınız.

Evet

topic

Kafka konusu (topic). Bir konu, Kafka'nın işlediği mesaj akışlarının bir toplamıdır.

Evet

column

Okunacak Kafka verileri. Sabit sütunlar, veri sütunları ve nitelik sütunları desteklenir.

• Sabit sütun: Tek tırnak içine alınmış bir sütundur, örneğin ["'abc'", "'123'"].

• Veri sütunu

  • Verileriniz JSON biçimindeyse, JSON nesnesinin özelliklerini alabilirsiniz, örneğin ["event_id"].

  • Verileriniz JSON biçimindeyse, JSON nesnesinin iç içe geçmiş alt özelliklerini alabilirsiniz, örneğin ["tag.desc"].

• Nitelik sütunu

  • __key__: Mesajın anahtarı.

  • __value__: Mesajın tam içeriği.

  • __partition__: Geçerli mesajın bulunduğu bölüm.

  • __headers__: Geçerli mesajın üstbilgileri.

  • __offset__: Geçerli mesajın ofseti.

  • __timestamp__: Geçerli mesajın zaman damgası.

  Aşağıdaki kod, eksiksiz bir örnek sunar.

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

Evet

keyType

Kafka anahtarının türü. Geçerli değerler: BYTEARRAY, DOUBLE, FLOAT, INTEGER, LONG ve SHORT.

Hayır

valueType

Kafka değerinin türü. Geçerli değerler: BYTEARRAY, DOUBLE, FLOAT, INTEGER, LONG ve SHORT.

Hayır

beginDateTime

Veri tüketimi için başlangıç zamanı. Bu parametre, dahil olan zaman aralığının sol sınırını belirtir. Değer, yyyymmddhhmmss biçiminde bir zaman dizesidir. Bu parametreyi scheduling parameters ile birlikte kullanabilirsiniz. Daha fazla bilgi için bkz. Zamanlama parametrelerinin desteklenen biçimleri.

Bu parametreyi veya beginOffset parametresini belirtmeniz gerekir.

endDateTime

Veri tüketimi için bitiş zamanı. Bu parametre, hariç tutulan zaman aralığının sağ sınırını belirtir. Değer, yyyymmddhhmmss biçiminde bir zaman dizesidir. Bu parametreyi scheduling parameters ile birlikte kullanabilirsiniz. Daha fazla bilgi için bkz. Zamanlama parametrelerinin desteklenen biçimleri.

Bu parametreyi veya endOffset parametresini belirtmeniz gerekir.

beginOffset

Veri tüketimi için başlangıç ofseti. Aşağıdaki biçimlerde yapılandırabilirsiniz:

• Tüketim için başlangıç ofsetini belirten bir sayı, örneğin 15553274.

• seekToBeginning: Verilerin en eski ofsetten tüketileceğini belirtir.

• seekToLast: kafkaConfig parametresinde belirtilen group.id için kaydedilen ofsetten başlayarak verileri okur. Grup ofsetinin istemci tarafından düzenli aralıklarla Kafka sunucusuna otomatik olarak işlendiğini unutmayın. Bu nedenle, bir görev başarısız olur ve yeniden çalıştırılırsa veri çoğaltması veya kaybı oluşabilir. skipExceedRecord parametresi true olarak ayarlanırsa görev, okunan son birkaç kaydı atabilir. Atılan bu veriler için grup ofseti zaten sunucuya işlendiğinden, bu veriler sonraki görev çalıştırmasında okunamaz.

• seekToEnd: Verilerin en son ofsetten tüketileceğini belirtir. Bu işlem boş veri okur.

Bu parametreyi veya beginDateTime parametresini belirtmeniz gerekir.

endOffset

Veri tüketimi için bitiş ofseti. Veri tüketimi görevinin ne zaman çıkacağını kontrol etmek için kullanılır.

Bu parametreyi veya endDateTime parametresini belirtmeniz gerekir.

skipExceedRecord

Kafka, veri tüketmek için public ConsumerRecords<K, V> poll(final Duration timeout) yöntemini kullanır. Tek bir poll çağrısı, belirtilen endOffset veya endDateTime değerinin ötesinde veri getirebilir. Bu parametre, bu aşırı verilerin hedefe yazılıp yazılmayacağını belirler. Görev otomatik ofset işleme kullandığından aşağıdakileri öneririz:

• 0.10.2'den önceki Kafka sürümleri için: skipExceedRecord değerini false olarak ayarlayın.

• Kafka 0.10.2 ve sonraki sürümler için: skipExceedRecord değerini true olarak ayarlayın.

Hayır. Varsayılan değer false değeridir.

partition

Bir Kafka konusunun birden çok bölümü (partition) vardır. Varsayılan olarak, bir veri senkronizasyon görevi, konudaki tüm bölümleri kapsayan bir ofset aralığından veri okur. Yalnızca tek bir bölümün ofset aralığından veri okumak için bir partition da belirtebilirsiniz.

Hayır. Varsayılan değer yok.

kafkaConfig

Veri tüketimi için bir KafkaConsumer istemcisi oluştururken bootstrap.servers, auto.commit.interval.ms ve session.timeout.ms gibi genişletilmiş parametreler belirtebilirsiniz. KafkaConsumer'ın tüketim davranışını kontrol etmek için kafkaConfig kullanabilirsiniz.

Hayır

encoding

keyType veya valueType STRING olarak ayarlandığında, dizeyi ayrıştırmak için bu parametreyle belirtilen kodlama kullanılır.

Hayır. Varsayılan değer UTF-8'dir.

waitTIme

Tüketici nesnesinin tek bir denemede Kafka'dan veri çekmek için beklediği maksimum süre (saniye cinsinden).

Hayır. Varsayılan değer 60'tır.

stopWhenPollEmpty

Geçerli değerler true ve false'tur. Bu parametre true olarak ayarlanırsa ve tüketici Kafka'dan boş veri çekerse (genellikle konudaki tüm veriler okunduğu için veya ağ ya da Kafka kümesi kullanılabilirlik sorunları nedeniyle), görev hemen durur. Aksi takdirde, tekrar veri okunana kadar yeniden dener.

Hayır. Varsayılan değer true'dur.

stopWhenReachEndOffset

Bu parametre yalnızca stopWhenPollEmpty true olduğunda etkin olur. Geçerli değerler true ve false'tur.

• Bu parametre true olarak ayarlanırsa ve tüketici bir poll isteğinden veri almazsa, konu bölümlerindeki en son ofsete ulaşılıp ulaşılmadığını kontrol eder. Tüm bölümler için en son ofsete ulaşıldıysa görev hemen durur. Aksi takdirde, konudan veri çekmeye devam etmeyi dener.

• Bu parametre false olarak ayarlanırsa ve tüketici bir poll isteğinden veri almazsa kontrol yapmaz ve görevi hemen durdurur.

Hayır. Varsayılan değer false'tur.

Parametre

Açıklama

fetch.min.bytes

Tüketicinin tek bir istekte brokerdan aldığı minimum veri miktarı (bayt cinsinden). Broker, tüketiciye yanıt vermeden önce bu miktarda verinin mevcut olmasını bekler.

fetch.max.wait.ms

Broker'ın bir getirme isteğine yanıt vermeden önce verilerin mevcut olması için beklediği maksimum süre (milisaniye cinsinden). Varsayılan değer 500'dür. Broker, fetch.min.bytes veya fetch.max.wait.ms koşullarından biri karşılandığında yanıt verir.

max.partition.fetch.bytes

Broker'ın her partition bölümünden tüketiciye döndürebileceği maksimum bayt sayısını belirtir. Varsayılan değer 1 MB'tır.

session.timeout.ms

Tüketicinin hizmet almayı bırakmadan önce sunucuyla bağlantısının kesilebileceği süreyi belirtir. Varsayılan değer 30 saniyedir.

auto.offset.reset

Tüketicinin ofset olmadan veya geçersiz bir ofsetle okuma yaptığında (tüketici uzun süre etkin olmadığı ve ofsete sahip kayıt süresi dolup silindiği için) gerçekleştirdiği eylem. Varsayılan değer none değeridir; bu, ofsetin otomatik olarak sıfırlanmadığı anlamına gelir. Bunu earliest olarak değiştirebilirsiniz; bu, tüketicinin partition kayıtlarını en eski ofsetten okuduğu anlamına gelir.

max.poll.records

Poll yönteminin tek bir çağrısında döndürülebilecek mesaj sayısı.

key.deserializer

Mesaj anahtarının seri durumdan çıkarma yöntemi, örneğin org.apache.kafka.common.serialization.StringDeserializer.

value.deserializer

Veri değerinin seri durumdan çıkarma yöntemi, örneğin org.apache.kafka.common.serialization.StringDeserializer.

ssl.truststore.location

SSL kök sertifikasının yolu.

ssl.truststore.password

Kök sertifika deposunun parolası. Alibaba Cloud Kafka kullanıyorsanız bunu KafkaOnsClient olarak ayarlayın.

security.protocol

Erişim protokolü. Şu anda yalnızca SASL_SSL protokolü desteklenmektedir.

sasl.mechanism

SASL kimlik doğrulama yöntemi. Alibaba Cloud Kafka kullanıyorsanız PLAIN kullanın.

java.security.auth.login.config

SASL kimlik doğrulama dosyasının yolu.

Parametre

Açıklama

Gerekli

datasource

Veri kaynağının adı. Kod düzenleyici, veri kaynakları eklemeyi destekler. Bu parametrenin değeri, eklenen veri kaynağının adıyla aynı olmalıdır.

Evet

server

Kafka sunucusunun adresi, ip:port biçimindedir.

Evet

topic

Kafka konusu (topic). Kafka'nın işlediği farklı mesaj akışları için bir kategoridir.

Bir Kafka kümesine yayınlanan her mesajın konu (topic) adı verilen bir kategorisi vardır. Bir konu, bir grup mesajın koleksiyonudur.

Evet

valueIndex

Kafka yazıcısında değer olarak kullanılan sütun. Bu belirtilmezse, varsayılan olarak tüm sütunlar birleştirilerek değer oluşturulur. Ayırıcı, fieldDelimiter tarafından belirtilir.

Hayır

writeMode

valueIndex yapılandırılmadığında, bu parametre kaynak kaydının tüm sütunlarını birleştirerek Kafka kaydının değerini oluşturmak için kullanılan biçimi belirler. Geçerli değerler text ve JSON'dur. Varsayılan değer text'tir.

• text olarak ayarlanırsa, tüm sütunlar fieldDelimiter tarafından belirtilen ayırıcı kullanılarak birleştirilir.

• JSON olarak ayarlanırsa, tüm sütunlar column parametresi tarafından belirtilen alan adlarına göre bir JSON dizesine birleştirilir.

Örneğin, bir kaynak kaydın a, b ve c değerlerine sahip üç sütunu varsa ve writeMode text olarak, fieldDelimiter ise # olarak ayarlanırsa, yazılan Kafka kaydının değeri a#b#c dizesi olur. writeMode JSON olarak ve column [{"name":"col1"},{"name":"col2"},{"name":"col3"}] olarak ayarlanırsa, yazılan Kafka kaydının değeri {"col1":"a","col2":"b","col3":"c"} dizesi olur.

valueIndex yapılandırılırsa bu parametre geçersizdir.

Hayır

column

Verilerin yazılacağı hedef tablodaki alanlar, virgülle ayrılmıştır. Örneğin: "column": ["id", "name", "age"].

valueIndex yapılandırılmadığında ve writeMode JSON olarak ayarlandığında, bu parametre kaynak kaydının sütun değerleri için JSON yapısındaki alan adlarını tanımlar. Örneğin, "column": [{"name":id","type":"JSON_NUMBER"}, {"name":"name","type":"JSON_STRING"}, {"name":"age","type":"JSON_NUMBER"}].

• Kaynak kayıttaki sütun sayısı, column içinde yapılandırılan alan adı sayısından fazlaysa yazma sırasında veriler kırpılır. Örneğin:

  Bir kaynak kaydın a, b ve c değerlerine sahip üç sütunu varsa ve column [{"name":"col1","type":"JSON_STRING"},{"name":"col2","type":"JSON_STRING"}] olarak yapılandırılırsa, yazılan Kafka kaydının değeri {"col1":"a","col2":"b"} dizesi olur.

• Kaynak kayıttaki sütun sayısı, column içinde belirtilen alan sayısından azsa ekstra alanlar null veya nullValueFormat tarafından belirtilen dizeyle doldurulur. Örneğin:

  Bir kaynak kaydın a ve b değerlerine sahip iki sütunu varsa ve column [{"name":"col1","type":"JSON_STRING"},{"name":"col2","type":"JSON_STRING"},{"name":"col3","type":"JSON_STRING"}] olarak yapılandırılırsa, yazılan Kafka kaydının değeri {"col1":"a","col2":"b","col3":null} dizesi olur. valueIndex yapılandırılırsa veya writeMode text olarak ayarlanırsa bu parametre geçersizdir.

• JSON alan türü yapılandırılmazsa varsayılan alan türü JSON_STRING'dir.

• JSON alan türünün geçerli değerleri için bkz. Ek: JSON alan türleri.

valueIndex yapılandırılırsa veya writeMode text olarak ayarlanırsa bu parametre geçersizdir.

valueIndex yapılandırılmadığında ve writeMode JSON olarak ayarlandığında gereklidir.

partition

Verilerin yazıldığı Kafka konusundaki bölüm numarasını belirtir. Bu, 0'a eşit veya 0'dan büyük bir tamsayı olmalıdır.

Hayır

keyIndex

Kafka yazıcısında anahtar olarak kullanılan sütun.

keyIndex parametresinin değeri 0'a eşit veya 0'dan büyük bir tamsayı olmalıdır. Aksi takdirde görev başarısız olur.

Hayır

keyIndexes

Kaynak kayıttaki, Kafka kaydı için anahtar olarak kullanılan sütunların sıra numaralarından oluşan bir dizi.

Sütun sıra numarası 0'dan başlar. Örneğin, [0,1,2] yapılandırılan tüm sütun numaralarının değerlerini virgülle birleştirerek Kafka kaydının anahtarını oluşturur. Bu belirtilmezse Kafka kaydının anahtarı null olur ve veriler konunun bölümlerine round-robin yöntemiyle yazılır. Yalnızca bu parametreyi veya keyIndex parametresini belirtebilirsiniz.

Hayır

fieldDelimiter

writeMode text olarak ayarlandığında ve valueIndex yapılandırılmadığında, kaynak kaydının tüm sütunları bu parametreyle belirtilen sütun ayırıcısı kullanılarak birleştirilerek Kafka kaydının değeri oluşturulur. Ayırıcı olarak tek bir karakter veya birden çok karakter yapılandırabilirsiniz. \u0001 biçiminde Unicode karakterleri yapılandırabilirsiniz. \t ve \n gibi kaçış karakterleri desteklenir. Varsayılan değer \t değeridir.

writeMode text olarak ayarlanmazsa veya valueIndex yapılandırılırsa bu parametre geçersizdir.

Hayır

keyType

Kafka anahtarının türü. Geçerli değerler: BYTEARRAY, DOUBLE, FLOAT, INTEGER, LONG ve SHORT.

Evet

valueType

Kafka değerinin türü. Geçerli değerler: BYTEARRAY, DOUBLE, FLOAT, INTEGER, LONG ve SHORT.

Evet

nullKeyFormat

keyIndex veya keyIndexes tarafından belirtilen kaynak sütunun değeri null ise bu değer, bu parametreyle belirtilen dizeyle değiştirilir. Bu yapılandırılmazsa değiştirme yapılmaz.

Hayır

nullValueFormat

Bir kaynak sütunun değeri null ise Kafka kaydının değeri oluşturulurken bu değer, bu parametreyle belirtilen dizeyle değiştirilir. Bu parametreyi belirtmezseniz değiştirme yapılmaz.

Hayır

acks

Kafka üreticisini başlatırken kullanılan acks yapılandırması. Bu, başarılı yazmalar için onaylama yöntemini belirler. Varsayılan olarak acks parametresi all olarak ayarlanır. acks için geçerli değerler şunlardır:

• 0: Başarılı yazmalar için onaylama yok.

• 1: Birincil çoğaltmaya başarılı yazma için onaylama.

• all: Tüm çoğaltmalara başarılı yazma için onaylama.

Hayır