The content of this page has been automatically translated by AI. If you encounter any problems while reading, you can view the corresponding content in Chinese.
2. Both Kafka and CKafka can select Kafka data source and follow the Kafka steps for read/write configuration.
Use Limits
1. Kafka message formats currently only support JSON (no nested), CSV (single delimiter), and Avro. Other special formats of JSON such as Oracle ogg json are not supported.
2. When both parameter.groupId and parameter.kafkaConfig.group.id are configured, parameter.groupId takes precedence over the group.id in kafkaConfig configuration.
Kafka Offline Single Table Read Node Configuration
Parameters
Description
Data Source
The data source type supported for Kafka reading is Kafka and CKafka.
topic
Kafka's Topic is an aggregation of feeds of messages that Kafka processes.
Serialization Format
Kafka data to be read supports constant columns, data columns, and attribute columns:
Constant Column: A column wrapped in single quotes is a constant column, for example, ["'abc'", "'123'"]
Data column
If your data is in JSON format, it supports fetching JSON attributes, e.g., ["event_id"]
If your data is in JSON format, it supports fetching nested JSON attributes, e.g., ["tag.desc"]
Attribute Column
__key__ represents the message key
__value__ represents the complete content of the message
__partition__ represents the partition of the current message
__headers__ represents the headers of the current message
__offset__ represents the offset of the current message
__timestamp__ represents the timestamp of the current message
Consumer Group ID
Avoid this parameter conflicting with other consumption processes to ensure the correctness of the consumption location. If not specified, the default setting is group.id=WeData_group_＄{Task ID}.
Cycle Start Point
During task cycle runtime, each read of Kafka's start point. The default is the scheduled time of the last cycle, optional: Partition Start Point, Consumer Group Current Point, Specified Locus, Specified Time.
Specified Time: When Kafka writes data, it automatically generates a unixtime timestamp as the time record of the data. The synchronization task reads the corresponding data from Kafka by converting the user's configured yyyymmddhhmmss value to a unixtimestamp, e.g., "beginDateTime": "20210125000000".
Partition Start Point: Extract data from the smallest undeleted offset in each partition of the Kafka topic.
Consumer Group Current Point: Read data starting from the specified offset saved by the consumer group ID configured in the task, usually the last stopping point of the process using this consumer group ID (ensure that this consumer group ID is only used by the configured DataInLong task to avoid data loss due to sharing the group ID). If using the group current point, must configure the consumer group ID; otherwise, the DataInLong task will randomly generate a group ID, and due to reset policies differences of the offset will cause errors or read from the start or end point. Group offsets are periodically auto-committed to Kafka server-side, so after a task failure, rerunning the task may lead to data duplication or loss. In wizard mode, records beyond the end point are automatically discarded, and these discarded records’ group offsets have already been committed to the server, making them unreadable in the next cycle.
Cycle End Point
During task cycle runtime, each read end point of Kafka. The default is the scheduled time of this cycle. When the keyType or valueType is configured as STRING, this option specifies the encoding for parsing the string.
Point Read Mode
The starting point for synchronizing data when the synchronization task is manually run. Provides two read modes:
latest: Read from the last offset position.
earliest: Read from the start point.
Kafka offline single table write node configuration
Parameters
Description
Data Flow Direction
The data source type supported for Kafka reading is Kafka and CKafka.
Topic
Kafka's Topic is an aggregation of feeds of messages that Kafka processes.
Serialization Format
Kafka data to be read supports constant columns, data columns, and attribute columns:
Constant Column: A column wrapped in single quotes is a constant column, for example, ["'abc'", "'123'"]
Data column:
If your data is in JSON format, it supports fetching JSON attributes, e.g., ["event_id"]
If your data is in JSON format, it supports fetching nested JSON attributes, e.g., ["tag.desc"]
Attribute column:
__key__ represents the message key.
__value__ represents the full content of the message.
__partition__ represents the partition of the current message.
__headers__ represents the headers of the current message.
__offset__ represents the offset of the current message.
__timestamp__ represents the timestamp of the current message.
Partition Partition Mapping
Support for polling write to partition, hash write based on specified fields, and specify partition modes.
If you choose the hash write based on specified fields mode, you need to specify the field name.
If you choose the specify partition mode, you need to set the partition number.
Advanced Settings (optional)
You can configure parameters based on business requirements.
Data type conversion is supported
Read
Other data types are converted to string type by default.
Format of reading Kafka value
Source field type
Internal Type
JSON
Integer、Long、BigInteger
Long
Float、Double
Double
Boolean
Bool
JSONArray、JSONObject
String
CSV
Integer、Long、BigInteger
Long
Float、Double
Double
Boolean
Bool
JSONArray、JSONObject
String
AVRO
Integer、Long、BigInteger
Long
Float、Double
Double
Boolean
Bool
JSONArray、JSONObject
String
Write
Other data types are converted to string type by default.
Format of writing Kafka value
Internal Type
Target end Field Type
JSON
Long
Integer、Bigint
Double
Float、Double、Decimal
Bool
Boolean
String
String、Varchar、Array
Date
Date、Timestamp
CSV
Long
Integer、Bigint
Double
Float、Double、Decimal
Bool
Boolean
String
String、Varchar、Array
Date
Date、Timestamp
AVRO
Long
Integer、Bigint
Double
Float、Double、Decimal
Bool
Boolean
String
String、Varchar、Array
Date
Date、Timestamp
Kafka/CKafka script demo
If you configure offline tasks using script mode, you need to write the reader parameters and writer parameters in the task script according to the unified script format requirements.
"job":{
"content":[
{
"reader":{
"parameter":{
"server":"ip:9092",
"endOffset":"9223372036854775807",//Cycle End Point
"kafkaConfig":{
"group.id":"WeData_group_1",//Consumer Group ID
"client.id":"WeData_client"
},
"messageFormat":"json",//serialization format
"column":[
"__key__",
"__value__",
"__partition__",
"__headers__",
"__offset__",
"__timestamp__"
],
"encoding":"UTF-8",
"stopWhenPollEmpty":true,
"beginOffset":"seekToLast",//cycle starting point
"valueType":"STRING",
"skipExceedRecord":"true",
"topic":"source_topic",//source topic
"keyType":"STRING",
"waitTime":120
},
"name":"kafkareader"
},
"transformer":[],
"writer":{
"parameter":{
"server":"ip:9092",
"messageFormat":"json",//serialization format
"kafkaConfig":{
"client.id":"WeData_client"
},
"partitioner":"roundRobin",// partition mapping method, supports polling, specify field hash, and specify partition
"column":[
{
"name":"key",
"type":"string"
},
{
"name":"value",
"type":"string"
},
{
"name":"partition",
"type":"string"
},
{
"name":"headers",
"type":"string"
},
{
"name":"offset",
"type":"string"
},
{
"name":"timestamp",
"type":"string"
}
],
"topic":"sink_topic"// target topic
},
"name":"kafkawriter"
}
}
],
"setting":{
"errorLimit":{// Dirty data threshold
"record":0
},
"speed":{
"byte":-1,// No limit on sync speed, positive integer indicates maximum transfer speed in byte/s
