# 配置Kafka源端参数
表1Kafka作为源端时的作业参数 
| 参数类型                                                 | 参数                                                           | 说明                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                              | 取值样例                |
|:---|:---|:---|:---|
| 基本参数  | 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示例   
     ![](https://support.huaweicloud.com/usermanual-dataartsstudio/zh-cn_image_0000002490702441.png "点击放大") 
    
2. 配置读取kafka任务。 将"消费记录策略"设置为"起止时间"，将"起始时间"设置为变量${startTime}，"结束时间"设置为变量${endTime}。
   图2配置消费记录策略   
   ![](https://support.huaweicloud.com/usermanual-dataartsstudio/zh-cn_image_0000002457503232.png "点击放大") 
3. 在执行调度中将任务设置为每小时调度。