From Kafka

**Table 1** Parameter description
Type	Parameter	Description	Example Value
Basic parameters	Topic	Topic name. You can add a single topic.	cdm_topic
	Data Format	Format used for parsing data. The options are as follows: JSON: Source data will be migrated after being converted in JSON format. CSV: Source data will be migrated after being converted in CSV format.	JSON
	Consumer Group ID	Consumer group ID If you export data from DMS Kafka, enter any value for Kafka Platinum but a valid consumer group ID for Kafka Basic.	consumer-group
	Consumption Record Policy	Consumption record policy Start/End Time: Check whether the extracted records are within the specified start time and end time based on the Kafka record metadata TIMESTAMP. When the end time of the consumed record arrives, the extraction task is terminated. The start time is included and end time is excluded. It can be used together with scheduling tasks. Earliest: Data is consumed from the start point. Latest: Data is consumed from the last point. Submitted indicates that data has been submitted. The start/end time, waiting time, and maximum extraction time are independent of each other. If any of the conditions is met, the Kafka extraction ends.	Start/End Time
	Minimum Timestamp	This parameter is mandatory if the Consumption Record Policy is set to Start/End Time. The format is yyyy-MM-dd HH:mm:ss. This parameter can be used together with variables in DataArts Factory.	2024-07-25 00:00:00
	End Time	This parameter is mandatory if the Consumption Record Policy is set to Start/End Time. The format is yyyy-MM-dd HH:mm:ss. This parameter can be used together with variables in DataArts Factory.	2024-07-25 23:59:59
	Wait Data Timeout	Time (in minutes) to wait before the task stops when null is returned for a data acquisition request from the consumer	30s
	Pull Data Runtime	Maximum extraction time for the consumer, in minutes It indicates the maximum duration for Kafka to extract data from the consumer end. When the duration ends, the extraction is forcibly terminated. If this parameter is not set, the default value 30 min is used.	1440
	Field Delimiter	Field delimiter during migration. The default value is a space.	,
	Record Delimiter	Currently, the following special characters cannot be used as separators: @\|\|$.	,

Example

You can use the start and end policies of Kafka together with the scheduling capability to implement periodic incremental synchronization. You can enable incremental data extraction from Kafka by hour or day using the start and end policies and periodic scheduling.

For example, if you want to write newly generated Kafka data to Hive every hour to implement periodic incremental data ingestion into the lake, perform the following operations:

Configure job parameter variables.
- Start time: startTime:#{DateUtil.format(DateUtil.addHours(Job.planTime,-1),"yyyy-MM-dd HH:mm:ss")}
- End time: endTime:#{DateUtil.format(Job.planTime,"yyyy-MM-dd HH:mm:ss")}
  startTime indicates the one hour before the task scheduling time, and endTime indicates the scheduling time.
  
  Figure 1 Example
Configure the task for reading Kafka data.
Set Consumption Record Policy to Time Range, Start Time to ${startTime}, and End Time to ${endTime}.

Figure 2 Configuring a consumption record policy
Set the task to be executed every hour.