DataWorks:Kafka data source

Data module

Data type after processing

key

The data type depends on the keyType parameter that you configure for a synchronization task. For more information about the keyType parameter, see the Parameters in code for Kafka Reader section in this topic.

value

The data type depends on the valueType parameter that you configure for a synchronization task. For more information about the valueType parameter, see the Parameters in code for Kafka Reader section in this topic.

offset

Long.

timestamp

Long.

headers

String.

partition

Long.

Configuration item

Description

Logon module class

Configure org.apache.kafka.common.security.plain.PlainLoginModule as the logon module class.

username

The username. Specify this configuration item based on your business requirements.

password

The password. Specify this configuration item based on your business requirements.

Parameter

Description

Required

datasource

The name of the data source. It must be the same as the name of the added data source. You can add data sources by using the code editor.

Yes.

server

The address of a Kafka broker in your Kafka cluster. Specify the address in the following format: IP address:Port number.

You can specify one or more broker addresses in the server parameter. You must make sure that DataWorks can use the specified addresses to access the related brokers.

Yes.

topic

The name of the Kafka topic from which you want to read data. Kafka maintains feeds of messages in topics.

Yes.

column

The names of the columns from which you want to read data. Constant columns, data columns, and property columns are supported.

• Constant column: a column whose name is enclosed in single quotation marks ('). Example: ["'abc'", "'123'"].

• Data column:

  • If your data is in the JSON format, you can obtain JSON properties. Example: ["event_id"].

  • If your data is in the JSON format, you can obtain the properties of nested objects in the data. Example: ["tag.desc"].

• Property column:

  • __key__: the key of a Kafka record.

  • __value__: the complete content of a Kafka record.

  • __partition__: the partition where a Kafka record resides.

  • __headers__: the header of a Kafka record.

  • __offset__: the offset of a Kafka record.

  • __timestamp__: the timestamp of a Kafka record.

  The following code provides a configuration example of the column parameter:

  "column": [
      "__key__",
      "__value__",
      "__partition__",
      "__offset__",
      "__timestamp__",
      "'123'",
      "event_id",
      "tag.desc"
      ]

Yes.

keyType

The data type of the key in the Kafka topic. Valid values: BYTEARRAY, DOUBLE, FLOAT, INTEGER, LONG, and SHORT.

Yes.

valueType

The data type of the value in the Kafka topic. Valid values: BYTEARRAY, DOUBLE, FLOAT, INTEGER, LONG, and SHORT.

Yes.

beginDateTime

The start time of data consumption. This parameter specifies the left boundary of a left-closed, right-open interval. Specify the time in the yyyymmddhhmmss format. This parameter can be used together with the scheduling parameters in DataWorks. For more information, see Supported formats of scheduling parameters.

You must configure either the beginDateTime or beginOffset parameter.

endDateTime

The end time of data consumption. This parameter specifies the right boundary of a left-closed, right-open interval. Specify the time in the yyyymmddhhmmss format. This parameter can be used together with the scheduling parameters in DataWorks. For more information, see Supported formats of scheduling parameters.

You must configure either the endDateTime or endOffset parameter.

beginOffset

The offset from which data consumption starts. The following formats are supported:

• Numeric string: Data consumption starts from the specified offset, such as 15553274.

• seekToBeginning: Data is consumed from the start offset.

• seekToLast: Data is consumed from the offset of the consumer group that is specified by the group.id parameter in kafkaConfig. The offset of the consumer group is automatically committed to the Kafka server on a regular basis. If a synchronization task fails and is rerun, data may be duplicated or lost. In this case, if you set the skipExceedRecord parameter to true, some data records that were last read may be discarded. As a result, the synchronization task cannot read the discarded data in the next cycle because the offset of the consumer group for the discarded data is already committed to the Kafka server.

• seekToEnd: Data is consumed from the end offset. In this case, null may be read.

You must configure either the beginOffset or beginDateTime parameter.

endOffset

The offset at which data consumption ends.

You must configure either the endOffset or endDateTime parameter.

skipExceedRecord

The Kafka consumer uses public ConsumerRecords<K, V> poll(final Duration timeout) to poll for data. However, the data obtained in a poll may be beyond the boundary that is specified by the endOffset or endDateTime parameter. The skipExceedRecord parameter specifies whether to poll for the excess data. Kafka Reader automatically commits offsets for data consumption. Therefore, we recommend that you configure the skipExceedRecord parameter based on the following instructions:

• If the data source from which you want to read data is of a version earlier than Kafka 0.10.2, set the skipExceedRecord parameter to false.

• If the data source from which you want to read data is of Kafka 0.10.2 or later, set the skipExceedRecord parameter to true.

No. Default value: false.

partition

If a Kafka topic contains multiple partitions, Kafka Reader reads data in a specific offset interval from the partitions. If you want Kafka Reader to read data from a specific partition, you can use the partition parameter to specify the partition.

No. This parameter does not have a default value.

kafkaConfig

The extended parameters that are specified when you create the Kafka consumer, such as bootstrap.servers, auto.commit.interval.ms, and session.timeout.ms. You can configure the parameters in kafkaConfig to manage the data consumption of the Kafka consumer.

No.

encoding

If the keyType or valueType parameter is set to STRING, strings are parsed based on the value of the encoding parameter.

No. Default value: UTF-8.

waitTIme

The maximum duration during which the Kafka consumer waits to poll for data from Kafka each time. Unit: seconds.

No. Default value: 60.

stopWhenPollEmpty

You can set this parameter to true or false. If you set this parameter to true, the Kafka consumer may poll for null from Kafka because all data in the Kafka topic is polled or an error occurs on the network or Kafka cluster. If the Kafka consumer polls for null, the related synchronization task immediately stops. If you set this parameter to false and the Kafka consumer polls for null, the Kafka consumer attempts to poll for data until the data is obtained.

No. Default value: true.

stopWhenReachEndOffset

This parameter takes effect only when the stopWhenPollEmpty parameter is set to true. You can set this parameter to true or false.

• If you set this parameter to true, and if null is returned from Kafka when the Kafka consumer polls for data, the system checks whether the data at the latest offset in all partitions of the Kafka topic is read. If the data at the latest offset in all partitions of the Kafka topic is read, the related synchronization task immediately stops. Otherwise, the Kafka consumer attempts to poll for data until the data is obtained.

• If you set this parameter to false, and if null is returned form Kafka when the Kafka consumer polls for data, the system does not perform a check and the related synchronization task immediately stops.

No. Default value: false.

Parameter

Description

fetch.min.bytes

The minimum number of bytes that the consumer can obtain from the broker. The broker returns data to the consumer only after the number of bytes reaches the specified value.

fetch.max.wait.ms

The maximum duration during which the consumer waits for data from the broker. Default value: 500. Unit: milliseconds. The broker returns data to the consumer when one of the conditions that are specified by the fetch.min.bytes and fetch.max.wait.ms parameters is met.

max.partition.fetch.bytes

The maximum number of bytes in each partition that the broker returns to the consumer. Default value: 1. Unit: MB.

session.timeout.ms

The timeout period of a connection session between the consumer and the broker. If this limit is exceeded, the consumer can no longer receive data from the broker. Default value: 30. Unit: seconds.

auto.offset.reset

The handling method that is used when no offset is available or the offset is invalid. This occurs if the consumer times out or the record with the specified offset expires and is deleted. The default value of this parameter is none, which indicates that the offset cannot be automatically reset. You can set this parameter to earliest, which indicates that the consumer reads partition data from the start offset.

max.poll.records

The number of Kafka records that can be returned for a single poll.

key.deserializer

The method that is used to deserialize the Kafka record key, such as org.apache.kafka.common.serialization.StringDeserializer.

value.deserializer

The method that is used to deserialize the Kafka record value, such as org.apache.kafka.common.serialization.StringDeserializer.

ssl.truststore.location

The path of the SSL root certificate.

ssl.truststore.password

The password of the truststore in the SSL root certificate. If you use ApsaraMQ for Kafka, set this parameter to KafkaOnsClient.

security.protocol

The access protocol. Set this parameter to SASL_SSL.

sasl.mechanism

The Simple Authentication and Security Layer (SASL) authentication mode. If you use ApsaraMQ for Kafka, set this parameter to PLAIN.

java.security.auth.login.config

The path of the SASL authentication file.

Parameter

Description

Required

datasource

The name of the data source. It must be the same as the name of the added data source. You can add data sources by using the code editor.

Yes

server

The address of a Kafka broker in your Kafka cluster. Specify the address in the format of IP address:Port number.

Yes

topic

The name of the Kafka topic to which you want to write data. Topics are categories in which Kafka maintains feeds of messages.

Each message that is published to a Kafka cluster is assigned to a topic. Each topic contains a group of messages.

Yes

valueIndex

The sequence number of the column that is obtained from a reader and used as the value in the Kafka topic. If you leave this parameter empty, all columns obtained from the reader are concatenated by using the delimiter specified by the fieldDelimiter parameter to form the value.

No

writeMode

The write mode. If you leave the valueIndex parameter empty, you can use the writeMode parameter to specify the format in which Kafka Writer concatenates all columns obtained from the reader. The default value is text. Valid values:

• text: Kafka Writer uses the delimiter specified by the fieldDelimiter parameter to concatenate all columns obtained from the reader.

• JSON: Kafka Writer concatenates all columns obtained from the reader into a JSON string based on the column names specified by the column parameter.

For example, three columns are obtained from the reader, and the values in the three columns are a, b, and c. If you set the writeMode parameter to text and the fieldDelimiter parameter to a number sign (#), Kafka Writer concatenates the columns into the string a#b#c and writes this string to Kafka. If you set the writeMode parameter to JSON and the column parameter to [{"name":"col1"},{"name":"col2"},{"name":"col3"}], Kafka Writer concatenates the columns into the JSON string {"col1":"a","col2":"b","col3":"c"} and writes this JSON string to Kafka.

If you configure the valueIndex parameter, the writeMode parameter is invalid.

No

column

The names of the columns to which you want to write data. Separate the names with commas (,), such as "column": ["id", "name", "age"].

If you leave the valueIndex parameter empty and set the writeMode parameter to JSON, the column parameter determines the names of the columns in the JSON string into which the columns obtained from the reader are concatenated, such as "column": [{"name":id"}, {"name":"name"}, {"name":"age"}].

• If the number of columns that are obtained from the reader is greater than the number of columns that are specified in the column parameter, Kafka Writer truncates the columns that are obtained from the reader. Example:

  Three columns are obtained from the reader, and the values in the columns are a, b, and c. If the column parameter is set to [{"name":"col1"},{"name":"col2"}], Kafka Writer concatenates the columns into the JSON string {"col1":"a","col2":"b"} and writes this JSON string to Kafka.

• If the number of columns that are obtained from the reader is less than the number of columns that are specified in the column parameter, Kafka Writer sets the values of the excess columns in Kafka to null or the string that is specified by the nullValueFormat parameter. Example:

  Two columns are obtained from the reader, and the values in the columns are a and b. If the column parameter is set to [{"name":"col1"},{"name":"col2"},{"name":"col3"}], Kafka Writer concatenates the columns into the JSON string {"col1":"a","col2":"b","col3":null} and writes this JSON string to Kafka. If you configure the valueIndex parameter or set the writeMode parameter to text, the column parameter is invalid.

If you configure the valueIndex parameter or set the writeMode parameter to text, the column parameter is invalid.

Required if valueIndex is not configured and writeMode is set to JSON

partition

The ID of the partition to which you want to write data. The value of this parameter must be an integer that is greater than or equal to 0.

No

keyIndex

The sequence number of the column that is obtained from the reader and used as the key in the Kafka topic.

The value of this parameter must be an integer that is greater than or equal to 0. If you set this parameter to an integer less than 0, an error occurs.

No

keyIndexes

The sequence numbers of the columns that are obtained from the reader and used as the key in the Kafka topic.

The sequence numbers must start from 0 and be separated by commas (,), such as [0,1,2]. If you leave this parameter empty, the key in the Kafka topic is null, and data is written to each partition in the Kafka topic in turn. You can configure either the keyIndex or keyIndexes parameter.

No

fieldDelimiter

The column delimiter. If you set the writeMode parameter to text and leave the valueIndex parameter empty, Kafka Writer uses the column delimiter you specify for the fieldDelimiter parameter to concatenate all columns that are obtained from the reader to form the value. You can use a single character or multiple characters as the column delimiter. The characters can be Unicode characters (such as \u0001) or escape characters (such as \t or \n). Default value: \t.

If the writeMode parameter is not set to text or the valueIndex parameter is configured, the fieldDelimiter parameter is invalid.

No

keyType

The data type of the key in the Kafka topic. Valid values: BYTEARRAY, DOUBLE, FLOAT, INTEGER, LONG, and SHORT.

Yes

valueType

The data type of the value in the Kafka topic. Valid values: BYTEARRAY, DOUBLE, FLOAT, INTEGER, LONG, and SHORT.

Yes

nullKeyFormat

If the column specified in the keyIndex or keyIndexes parameter contains the value null, Kafka Writer replaces null with the value of the nullKeyFormat parameter. If you leave the nullKeyFormat parameter empty, Kafka Writer retains the value null.

No

nullValueFormat

If a column obtained from the reader contains the value null, Kafka Writer replaces null with the value of the nullValueFormat parameter. If you leave the nullValueFormat parameter empty, Kafka Writer retains the value null.

No

acks

The acknowledgment configuration used when the Kafka producer is initialized. This parameter specifies the method used to confirm that data is written to Kafka. Default value: all. Valid values:

• 0: A Kafka producer does not acknowledge whether data is written to the destination.

• 1: A Kafka producer acknowledges that the write operation is successful if data is written to the primary replica.

• all: A Kafka producer acknowledges that the write operation is successful if data is written to all replicas.

No