Help & Documentation>TDMQ for CKafka>Operation Guide>CKafka Connector>Introduction>Task Management>Data Processing Rule Description

Data Processing Rule Description

Last updated: 2025-07-18 11:49:21

On this page:

Overview

When handling data inflow and outflow tasks through the CKafka Connector, it is often necessary to perform simple data cleaning operations, such as formatting raw data, parsing specific fields, and converting data formats. Developers typically need to set up their own data cleaning (ETL) services.
Logstash is a free and open-source server-side data processing pipeline that can collect data from multiple sources, transform it, and then send it to the appropriate storage repositories. With its rich set of filter plugins, Logstash has become a widely used and powerful data transformation tool.
However, setting up, configuring, and maintaining your own Logstash service can increase the complexity of development and operations. To address this, CKafka offers a data processing service that mirrors Logstash's capabilities. Developers can easily create their own data processing tasks through the console's interactive interface. The data processing service allows users to edit corresponding data processing rules, supports building chain processing, and provides a preview of the data processing results. This helps users effortlessly and efficiently build a data processing service that meets their data cleansing and transformation needs.




List of Benchmarked Features

Logstash
Connector Data Processing Service
SDK
Codec.json
✔
Filter.grok
✔
Filter.mutate.split
✔
Filter.date
✔
Filter.json
✔
Filter.mutate.convert
✔
Filter.mutate.gsub
✔
Filter.mutate.strip
✔
Filter.mutate.join
✔
Filter.mutate.rename
✔
Filter.mutate.update
✔
Filter.mutate.replace
✔
Filter.mutate.add_field
✔
Filter.mutate.remove_field
✔
Filter.mutate.copy
✔
Filter.mutate.merge

TODO
Filter.mutate.uppercase

TODO
Filter.mutate.lowercase

TODO

Directions

Data Parsing

Logstash processing method:
  // Codec.json 
  input {
      file {
          path => "/var/log/nginx/access.log_json""
          codec => "json"
      }
  }
  // Filter.grok
  filter {
      grok {
          match => {
              "message" => "\s+(?<request_time>\d+(?:\.\d+)?)\s+"
          }
      }
  }
  // Filter.mutate.split
  filter {
      split {
          field => "message"
          terminator => "#"
      }
  }
Connector Processing Method: Select the appropriate data parsing mode and preview with a single click:


Date format processing

Logstash processing method:
  // Filter.date
  filter {
      date {
          match => ["client_time", "yyyy-MM-dd HH:mm:ss"] 
      }
  }
Connector Processing Methods:
1.1 Assign a value to a field by presetting the current system time:

1.2 Process data content using the Process Value feature:




Internal JSON structure parsing

Logstash processing method:
  // Filter.json
  filter {
      json {
          source => "message"
          target => "jsoncontent"
      }
  }
Connector Processing Method: Parse a specific field by selecting the MAP operation, which converts the field into JSON format:


Data Modification

Logstash processing method:
  // Filter.mutate.convert
  filter {
      mutate {
          convert => ["request_time", "float"]
      }
  }
  // Filter.mutate.gsub
  filter {
      mutate {
          gsub => ["urlparams", ",", "_"]
      }
  }

  // Filter.mutate.strip
  filter {
      mutate {
          strip => ["field1", "field2"]
      }
  }

  // Filter.mutate.join
  filter {
      mutate {
         join => { "fieldname" => "," }
       }
  }

Connector Processing Method: Quickly define a rule by selecting the corresponding Process Value option:



1.1 Quickly change the data format of a field by selecting the target data format:

1.2 Implement the join concatenation function using JSONPATH syntax, such as using
$.concat($.data.Response.SubnetSet[0].VpcId, \"#\", $.data.Response.SubnetSet[0].SubnetId, \"#\", $.data.Response.SubnetSet[0].CidrBlock)
to concatenate VPC and subnet attributes, separated by the # character.
1.3 The result is as follows:


Field modification

Logstash processing method:
  // Filter.mutate.rename
  filter {
      mutate {
          rename => ["syslog_host", "host"]
      }
  }
  // Filter.mutate.update
  filter {
      mutate {
          update => { "sample" => "My new message" }
       }
  }
  // Filter.mutate.replace
  filter {
      mutate {
          replace => { "message" => "%{source_host}: My new message" }
      }
  }
  // Filter.mutate.add_field
  filter {
      mutate {
          split => { "hostname" => "." }
          add_field => { "shortHostname" => "%{[hostname][0]}" }
      }
  }
  // Filter.mutate.remove_field
  filter {    
      mutate {            
          remove_field => ["field_name"]    
      }
   }
  // Filter.mutate.copy
  filter {
      mutate {
           copy => { "source_field" => "dest_field" }
      }
  }
Connector Processing Method:


Samples

Sample 1: Multi-level field parsing

Input message:
{
    "@timestamp": "2022-02-26T22:25:33.210Z",
    "beat": {
        "hostname": "test-server",
        "ip": "6.6.6.6",
        "version": "5.6.9"
    },
    "input_type": "log",
    "message": "{\"userId\":888,\"userName\":\"testUser\"}",
    "offset": 3030131,
}
Target message:
{
    "@timestamp": "2022-02-26T22:25:33.210Z",
    "input_type": "log",
    "hostname": "test-server",
    "ip": "6.6.6.6",
    "userId": 888,
    "userName": "testUser"
}
Connector Configuration Method:
1.1 Processing link 1 is configured as follows:

1.2 The result of processing link 1 is as follows:
{
    "@timestamp": "2022-02-26T22:25:33.210Z",
    "input_type": "log",
    "message": "{\"userId\":888,\"userName\":\"testUser\"}",
    "hostname": "test-server",
    "ip": "6.6.6.6"
}
1.3 Processing link 2 is configured as follows:

1.4 The result of processing link 2 is as follows:
{
    "@timestamp": "2022-02-26T22:25:33.210Z",
    "input_type": "log",
    "hostname": "test-server",
    "ip": "6.6.6.6",
    "userId": 888,
    "userName": "testUser"
}

Sample 2: Non-JSON data parsing

Input message:
region=Shanghai$area=a1$server=6.6.6.6$user=testUser$timeStamp=2022-02-26T22:25:33.210Z
Target message:
{
    "region": "Shanghai",
    "area": "a1",
    "server": "6.6.6.6",
    "user": "testUser",
    "timeStamp": "2022-02-27 06:25:33",
    "processTimeStamp": "2022-06-27 11:14:49"
}
Connector Configuration Method:
1.1 Parse the original message with the separator $.

1.2 Preliminary parsing result:
{
  "0": "region=Shanghai",
  "1": "area=a1",
  "2": "server=6.6.6.6",
  "3": "user=testUser",
  "4": "timeStamp=2022-02-26T22:25:33.210Z"
}
1.3 Perform a secondary parsing of the result using the separator =:

1.4 Secondary parsing result:
{
  "0": "region=Shanghai",
  "1": "area=a1",
  "2": "server=6.6.6.6",
  "3": "user=testUser",
  "4": "timeStamp=2022-02-26T22:25:33.210Z",
  "0.region": "Shanghai",
  "1.area": "a1",
  "2.server": "6.6.6.6",
  "3.user": "testUser",
  "4.timeStamp": "2022-02-26T22:25:33.210Z"
}
1.5 Edit and delete fields, adjust the timestamp format, and add the current system time field:


Final result:
{
    "region": "Shanghai",
    "area": "a1",
    "server": "6.6.6.6",
    "user": "testUser",
    "timeStamp": "2022-02-27 06:25:33",
    "processTimeStamp": "2022-06-27 11:14:49"
}

中国站 - English