文档首页/数据治理中心 DataArts Studio/用户指南/数据集成（离线作业）/配置作业源端参数/配置Kafka源端参数
更新时间：2025-11-18 GMT+08:00
查看PDF
分享

配置Kafka源端参数

表1 Kafka作为源端时的作业参数

参数类型

参数

说明

取值样例

基本参数

Topic

主题名称。支持单个topic。

cdm_topic

数据格式

解析数据时使用的格式：

  • JSON：以JSON格式解析源数据。
  • CSV格式：以CSV格式解析源数据。

JSON格式

消费组ID

用户指定消费组ID。

如果是从DMS Kafka导出数据，专享版请任意输入，标准版请输入有效的消费组ID。

consumer-group

消费记录策略

消费record策略。

  • 起止时间：根据kafka record元数据TIMESTAMP判断，抽取的record是否符合填入的起止时间范围。当消费到的record到达结束时间，则终止抽取任务。起止时间范围左闭右开:[起始时间，结束时间)。可配合调度任务使用。
  • 最早：表示从开始点位消费数据。
  • 最新：表示从最后点位消费数据。已提交：拉取已提交的数据。起止时间策略，等待时间，最大抽取时间相互独立。只要有任意一个条件符合，则kafka抽取结束。

起止时间

起始时间

消费记录策略为起始时间时须设置起始时间。

格式为yyyy-MM-dd HH:mm:ss，支持配合DLF变量等方式设置。

2024-07-25 00:00:00

结束时间

消费记录策略为起始时间时须设置结束时间。

格式为yyyy-MM-dd HH:mm:ss，支持配合DLF变量等方式设置。

2024-07-25 23:59:59

等待时间

消费者获取数据返回值为空，持续X秒，任务停止。

30秒

最大抽取时间

消费者最大抽取时间，单位min。

kafka抽取consumer端最大运行时间，当到达运行时间，抽取强制结束，如不填入，默认为30min。

1440

字段分隔符

迁移时的字段分割符，默认为空格。

,

记录分隔符

暂不支持@ || $特殊字符作为分隔符。

,

示例

kafka起止策略搭配调度能力可以完成周期增量同步。kafka常见按小时或天周期增量抽取数据，可以使用起止策略搭配周期调度实现。

比如用户希望每个小时将新生成的kafka数据，写入hive入湖，实现周期增量数据入湖，需进行以下操作。

  1. 配置作业参数变量。
    • 起始时间：startTime:#{DateUtil.format(DateUtil.addHours(Job.planTime,-1),"yyyy-MM-dd HH:mm:ss")}
    • 结束时间：endTime:#{DateUtil.format(Job.planTime,"yyyy-MM-dd HH:mm:ss")}

      startTime代表任务调度时间前一个小时，endTime代表调度时间。

      图1 示例
  2. 配置读取kafka任务。

    将“消费记录策略”设置为“起止时间”，将“起始时间”设置为变量${startTime}，“结束时间”设置为变量${endTime}。

    图2 配置消费记录策略
  3. 在执行调度中将任务设置为每小时调度。