Kafka記錄數據模塊

處理后的數據類型

key

取決于數據同步任務配置的keyType配置項，keyType參數介紹請參見下文的全量參數說明章節。

value

取決于數據同步任務配置的valueType配置項，valueType參數介紹請參見下文的全量參數說明章節。

offset

Long

timestamp

Long

headers

String

partition

Long

配置項

說明

登錄模塊

必須配置org.apache.kafka.common.security.plain.PlainLoginModul

username

用戶名，請根據實際情況配置該項。

password

密碼，請根據實際情況配置該項。

參數

描述

是否必選

datasource

數據源名稱，腳本模式支持添加數據源，此配置項填寫的內容必須要與添加的數據源名稱保持一致。

是

server

Kafka的broker server地址，格式為ip:port。

您可以只配置一個server，但請務必保證Kafka集群中所有broker的IP地址都可以連通DataWorks。

是

topic

Kafka的Topic，是Kafka處理資源的消息源（feeds of messages）的聚合。

是

column

需要讀取的Kafka數據，支持常量列、數據列和屬性列：

• 常量列：使用單引號包裹的列為常量列，例如["'abc'", "'123'"]。

• 數據列

  • 如果您的數據是一個JSON，支持獲取JSON的屬性，例如["event_id"]。

  • 如果您的數據是一個JSON，支持獲取JSON的嵌套子屬性，例如["tag.desc"]。

• 屬性列

  • __key__表示消息的key。

  • __value__表示消息的完整內容 。

  • __partition__表示當前消息所在分區。

  • __headers__表示當前消息headers信息。

  • __offset__表示當前消息的偏移量。

  • __timestamp__表示當前消息的時間戳。

  完整示例如下。

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

是

keyType

Kafka的Key的類型，包括BYTEARRAY、DOUBLE、FLOAT、INTEGER、LONG和SHORT。

是

valueType

Kafka的Value的類型，包括BYTEARRAY、DOUBLE、FLOAT、INTEGER、LONG和SHORT。

是

beginDateTime

數據消費的開始時間位點，為時間范圍（左閉右開）的左邊界。yyyymmddhhmmss格式的時間字符串，可以配合調度參數使用。詳情請參見調度參數支持的格式。

需要和beginOffset二選一。

endDateTime

數據消費的結束時間位點，為時間范圍（左閉右開）的右邊界。yyyymmddhhmmss格式的時間字符串，可以配合調度參數使用。詳情請參見調度參數支持的格式。

需要和endOffset二選一。

beginOffset

數據消費的開始時間位點，您可以配置以下形式：

• 數字形式（例如15553274），表示開始消費的點位。

• seekToBeginning：表示從開始點位消費數據。

• seekToLast：表示從kafkaConfig配置中指定的group.id對應群組ID保存的位點開始讀取數據，注意群組位點在客戶端會定時自動提交到Kafka服務端，所以任務失敗后，如果重跑任務時可能會有數據重復或者丟失，skipExceedRecord參數配置為true時，任務可能丟棄最后讀取的一些記錄，而這些丟棄數據的群組位點已經提交到服務端，在下一個周期任務運行時將無法讀到這些丟棄的數據。

• seekToEnd：表示從最后點位消費數據，會讀取到空數據。

需要和beginDateTime二選一。

endOffset

數據消費的結束位點，用于控制結束數據消費任務退出的時間。

需要和endDateTime二選一。

skipExceedRecord

Kafka使用public ConsumerRecords<K, V> poll(final Duration timeout)消費數據，一次poll調用獲取的數據可能在endOffset或者endDateTime之外。skipExceedRecord用于控制是否寫出多余的數據至目的端。由于消費數據使用了自動點位提交，建議您：

• Kafka 0.10.2之前版本：建議配置skipExceedRecord為false。

• Kafka 0.10.2及以上版本：建議配置skipExceedRecord為true。

否，默認值為false。

partition

Kafka的一個Topic有多個分區（partition），正常情況下數據同步任務是讀取Topic（多個分區）一個點位區間的數據。您也可以指定partition，僅讀取一個分區點位區間的數據。

否，無默認值。

kafkaConfig

創建Kafka數據消費客戶端KafkaConsumer可以指定擴展參數，例如bootstrap.servers、auto.commit.interval.ms、session.timeout.ms等，您可以基于kafkaConfig控制KafkaConsumer消費數據的行為。

否

encoding

當keyType或valueType配置為STRING時，將使用該配置項指定的編碼解析字符串。

否，默認為UTF-8。

waitTIme

消費者對象從Kafka拉取一次數據的最大等待時間，單位為秒。

否，默認為60。

stopWhenPollEmpty

該配置項可選值為true/false。當配置為true時，如果消費者從Kafka拉取數據返回為空（一般是已經讀完主題中的全部數據，也可能是網絡或者Kafka集群可用性問題），則立即停止任務，否則持續重試直到再次讀到數據。

否，默認為true。

stopWhenReachEndOffset

該配置項只在stopWhenPollEmpty為true時生效，可選值為true/false。

• 當配置為true時，如果消費者從Kafka拉取數據返回為空時，會檢查當前是否讀取到了Kafka Topic分區中的最新位點數據，如果已經讀到了Kafka Topic所有分區中的最新位點數據，則立即停止任務，否則繼續嘗試從Kafka Topic中拉取數據。

• 當配置為false時，如果消費者從Kafka拉取數據返回為空時，不會進行檢查，立即停止任務。

否，默認為false。

參數

描述

fetch.min.bytes

指定消費者從broker獲取消息的最小字節數，即有足夠的數據時，才將其返回給消費者。

fetch.max.wait.ms

等待broker返回數據的最大時間，默認500毫秒。fetch.min.bytes和fetch.max.wait.ms先滿足哪個條件，便按照該方式返回數據。

max.partition.fetch.bytes

指定broker從每個partition中返回給消費者的最大字節數，默認為1 MB。

session.timeout.ms

指定消費者不再接收服務之前，可以與服務器斷開連接的時間，默認是30秒。

auto.offset.reset

消費者在讀取沒有偏移量或者偏移量無效的情況下（因為消費者長時間失效，包含偏移量的記錄已經過時并被刪除）的處理方式。默認為none（意味著不會自動重置位點），您可以更改為earliest（消費者從起始位置讀取partition的記錄）。

max.poll.records

單次調用poll方法能夠返回的消息數量。

key.deserializer

消息Key的反序列化方法，例如org.apache.kafka.common.serialization.StringDeserializer。

value.deserializer

數據Value的反序列化方法，例如org.apache.kafka.common.serialization.StringDeserializer。

ssl.truststore.location

SSL根證書的路徑。

ssl.truststore.password

根證書Store的密碼。如果是Aliyun Kafka，則配置為KafkaOnsClient。

security.protocol

接入協議，目前支持使用SASL_SSL協議接入。

sasl.mechanism

SASL鑒權方式，如果是Aliyun Kafka，使用PLAIN。

java.security.auth.login.config

SASL鑒權文件路徑。

參數

描述

是否必選

datasource

數據源名稱，腳本模式支持添加數據源，此配置項填寫的內容必須要與添加的數據源名稱保持一致。

是

server

Kafka的server地址，格式為ip:port。

是

topic

Kafka的topic，是Kafka處理資源的消息源（feeds of messages）的不同分類。

每條發布至Kafka集群的消息都有一個類別，該類別被稱為topic，一個topic是對一組消息的歸納。

是

valueIndex

Kafka Writer中作為Value的那一列。如果不填寫，默認將所有列拼起來作為Value，分隔符為fieldDelimiter。

否

writeMode

當未配置valueIndex時，該配置項決定將源端讀取記錄的所有列拼接作為寫入kafka記錄Value的格式，可選值為text和JSON，默認值為text。

• 配置為text，將所有列按照fieldDelimiter指定分隔符拼接。

• 配置為JSON，將所有列按照column參數指定字段名稱拼接為JSON字符串。

例如源端記錄有三列，值為a、b和c，writeMode配置為text、fieldDelimiter配置為#時，寫入kafka的記錄Value為字符串a#b#c；writeMode配置為JSON、column配置為[{"name":"col1"},{"name":"col2"},{"name":"col3"}]時，寫入kafka的記錄Value為字符串{"col1":"a","col2":"b","col3":"c"}。

如果配置了valueIndex，該配置項無效。

否

column

目標表需要寫入數據的字段，字段間用英文逗號分隔。例如："column": ["id", "name", "age"]。

當未配置valueIndex，并且writeMode選擇JSON時，該配置項定義源端讀取記錄的列值在JSON結構中的字段名稱。例如，"column": [{"name":id"}, {"name":"name"}, {"name":"age"}]。

• 當源端讀取記錄列的個數多于column配置的字段名個數時，寫入時進行截斷。例如：

  源端記錄有三列，值為a、b和c，column配置為[{"name":"col1"},{"name":"col2"}]時，寫入kafka的記錄Value為字符串{"col1":"a","col2":"b"}。

• 當源端讀取記錄列的個數少于column配置的字段名個數時，多余column配置字段名填充null或者nullValueFormat指定的字符串。例如：

  源端記錄有兩列，值為a和b，column配置為[{"name":"col1"},{"name":"col2"},{"name":"col3"}]時，寫入kafka的記錄Value為字符串{"col1":"a","col2":"b","col3":null}。如果配置了valueIndex，或者writeMode配置為text，該配置項無效。

如果配置了valueIndex，或者writeMode配置為text，該配置項無效。

當未配置valueIndex，并且writeMode配置為JSON時必選

partition

指定寫入Kafka topic指定分區的編號，是一個大于等于0的整數。

否

keyIndex

Kafka Writer中作為Key的那一列。

keyIndex參數取值范圍是大于等于0的整數，否則任務會出錯。

否

keyIndexes

源端讀取記錄中作為寫入kafka記錄Key的列的序號數組。

列序號從0開始，例如[0,1,2]，會將配置的所有列序號的值用逗號連接作為寫入kafka記錄的Key。如果不填寫，寫入kafka記錄Key為null，數據輪流寫入topic的各個分區中，與keyIndex參數只能二選一。

否

fieldDelimiter

當writeMode配置為text，并且未配置valueIndex時，將源端讀取記錄的所有列按照該配置項指定列分隔符拼接作為寫入kafka記錄的Value，支持配置單個或者多個字符作為分隔符，支持以\u0001格式配置unicode字符，支持\t、\n等轉義字符。默認值為\t。

如果writeMode未配置為text或者配置了valueIndex，該配置項無效。

否

keyType

Kafka的Key的類型，包括BYTEARRAY、DOUBLE、FLOAT、INTEGER、LONG和SHORT。

是

valueType

Kafka的Value的類型，包括BYTEARRAY、DOUBLE、FLOAT、INTEGER、LONG和SHORT。

是

nullKeyFormat

keyIndex或者keyIndexes指定的源端列值為null時，替換為該配置項指定的字符串，如果不配置不做替換。

否

nullValueFormat

當源端列值為null時，組裝寫入kafka記錄Value時替換為該配置項指定的字符串，如果不配置不做替換。

否

acks

初始化Kafka Producer時的acks配置，決定寫入成功的確認方式。默認acks參數為all。acks取值如下：

• 0：不進行寫入成功確認。

• 1：確認主副本寫入成功。

• all：確認所有副本寫入成功。

否