Dataphin:Configure the Kafka input component

Procedure

1. In the top navigation bar of the Dataphin homepage, choose Develop > Data Integration.

2. In the top navigation bar of the integration page, select a project. In Dev-Prod mode, you must also select an environment.

3. In the left navigation bar, click Offline Integration. In the Offline Integration list, click the offline pipeline that you want to develop to open its configuration page.

4. In the upper-right corner of the page, click Component Library to open the Component Library panel.

5. In the left-side navigation pane of the Component Library panel, select Input. Find the Kafka component in the list on the right and drag it onto the canvas.

6. On the Kafka input component card, click the icon to open the Kafka Input Configuration dialog box.

7. In the Kafka Input Configuration dialog box, configure the parameters as described in the following table.

   Parameter	Description
   Step name	The name of the Kafka input component. Dataphin automatically generates a step name. You can also change it to suit your business needs. The name must adhere to the following rules: Can contain only Chinese characters, letters, underscores (_), and digits. Cannot exceed 64 characters in length.
   Data source	In the drop-down list, select a Kafka data source. The list displays all Kafka data sources in Dataphin, regardless of whether you have the read permission for synchronization. Click the icon to copy the current data source name. For a data source where you lack read permission for synchronization, click Apply next to the data source to request the permission. For more information, see Apply for data source permissions. If no Kafka data source is available, click Create Data Source to create one. For more information, see Create a Kafka data source.
   Topic	The Kafka topic. From the drop-down list, select the topic from which to read data.
   Key type	The type of the Kafka key. This setting determines the key.deserializer configuration when initializing the Kafka consumer. Valid values are BYTEARRAY, DOUBLE, FLOAT, INTEGER, LONG, SHORT, STRING, and Kafka Avro. You can select Kafka Avro only if schema.registry is configured for the data source.
   Value type	The type of the Kafka value. This setting determines the value.deserializer configuration when initializing the Kafka consumer. Valid values are BYTEARRAY, DOUBLE, FLOAT, INTEGER, LONG, SHORT, STRING, and Kafka Avro. You can select Kafka Avro only if schema.registry is configured for the data source.
   Value path (Optional)	If the Kafka message value is in JSON format and the business data is nested as an array within a specific field, you can use a value path to specify the array to parse. After you enter a JSONPath expression, the system expands the array into multiple rows, with one element per row. Metadata fields are repeated in each row. If you leave this parameter empty, each Kafka message is parsed as a single row. For example, if a Kafka message value has the structure {"__value__":{"datas":[...]}}, configure this parameter as follows: If datas is an array, set the path to $.__value__.datas[*]. This reads the datas array within __value__ and expands each element into a separate row. If datas is an object, set the path to $.__value__.datas.
   Consumer group ID	The group.id configuration for initializing the Kafka consumer. To ensure correct consumption offsets during synchronization, do not use a consumer group ID that is already used by other consumer processes. If you do not specify this parameter, a random string prefixed with datax_ is automatically generated as the group.id for each task.
   Start time	The start time for reading data. You must specify a specific time in yyyyMMddHHmmss format, which is the inclusive start of the time range. This parameter must be used with scheduling parameters. For example, if you set a scheduling parameter to beginDateTime=${20220101000000}, set Start Time to ${beginDateTime}.
   End time	The end time for reading data. You must specify a specific time in yyyyMMddHHmmss format, which is the inclusive end of the time range. This parameter must be used with scheduling parameters. For example, if you set a scheduling parameter to endDateTime=${20220101000000}, set End Time to ${endDateTime}.
   Synchronization stop policy	Select a policy to determine when the synchronization task stops. The following options are available: When no new data is read for 1 minute: If the consumer polls Kafka and receives an empty result for one minute, the task stops immediately. This typically occurs when all data in the topic has been read, but it can also be caused by network issues or problems with Kafka cluster availability. Otherwise, the task continues to retry until data is read. When the specified end offset is reached: The task stops when the business timestamp or offset of a read Kafka record matches the configured end criteria. Otherwise, the task retries indefinitely to read Kafka records.
   Advanced settings	You can use advanced settings to configure options such as the offset reset policy, maximum records per read, maximum read time, and read timeout. If the topic is configured with a schema registry, you must also configure the keySchema and valueSchema parameters in the advanced settings. This field is empty by default. The following code shows a sample configuration: { "namespace": "example.avro", "type": "record", "name": "User", "fields": [ {"name": "name", "type": "string"}, {"name": "favorite_number", "type": ["int", "null"]}, {"name": "favorite_color", "type": ["string", "null"]} ] }
   Output fields	By default, the following six fields are displayed: __key__, __value__, __partition__, __headers__, __offset__, and __timestamp__. You can also manually add output fields: Click Batch add to configure fields in batches using JSON, TEXT, or Sample Data format. JSON format example [ { "index": 0, "name": "__key__", "type": "STRING" }, { "index": 1, "name": "__value__", "type": "STRING" }, { "index": 2, "name": "__partition__", "type": "INTEGER" }, { "index": 3, "name": "__headers__", "type": "STRING" }, { "index": 4, "name": "__offset__", "type": "LONG" }, { "index": 5, "name": "__timestamp__", "type": "LONG" } ] Note index specifies the 0-based index of the source column, name specifies the field name after import, and type specifies the field type after import. For example, "index":3,"name":"user_id","type":"String" imports the 4th column from the source, names the field user_id, and sets its type to String. TEXT format example 0,__key__,STRING 1,__value__,STRING 2,__partition__,INTEGER 3,__headers__,STRING 4,__offset__,LONG 5,__timestamp__,LONG Note The row delimiter separates field entries. The default is the newline character (\n). The semicolon (;) and period (.) are also supported. The column delimiter separates the field name and the field type. The default is a comma (,). Select Sample Data, paste a sample JSON object into the text box, and click OK. The system automatically parses the field list at the corresponding level based on the Value path: If a Value path is configured, the system parses the fields at the level specified by the path. If the Value path is empty, the system parses the fields at the first level. After parsing is complete, the current output field list is overwritten. The field type defaults to STRING. Click Add Output Field and enter the Source Index and Field name, and select a Type as prompted. You can also configure source fields with strings other than the six default metadata fields. In this case, the Kafka record's value is parsed as a JSON string, and the configured source field is used as a JSON path to read the corresponding content. This value is then written to the corresponding destination field. For example: If a Kafka record's value is { "data": { "name": "bob", "age": 35 } } and you configure the source field as data.name, the value bob is extracted and written to the corresponding destination field. Supported field types include Java types and their corresponding DataX mappings. You can also perform the following actions on added fields: Click the icon in the Actions column to edit an existing field. Click the icon in the Actions column to delete an existing field.

8. Click OK to complete the configuration of the Kafka input component.