大数据开发治理平台 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：确认所有副本写入成功。

否