DataWorks：Kafka資料來源

參數

描述

是否必選

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：確認所有副本寫入成功。

否