Data cleansing rules-ApsaraMQ for Kafka(Kafka)-阿里云帮助中心

The data cleansing feature provides common templates for message processing, including content splitting, dynamic routing, content enrichment, and content mapping. You can use these templates to process messages out of the box or modify the code to meet your specific needs.

Background

A data cleansing task, powered by Function Compute, provides basic operator capabilities. The data cleansing feature supports the following products: ApsaraMQ for RocketMQ, ApsaraMQ for Kafka, ApsaraMQ for MQTT, ApsaraMQ for RabbitMQ, and Simple Message Queue (formerly MNS). After you create a data cleansing task, you can log on to the Function Compute console to customize the code and modify the function's configuration.

Operator	Description
Content splitting	Splits message content based on a regular expression and sends the resulting messages to the sink individually.
Dynamic routing	Routes matched messages to a specific sink and unmatched messages to a default sink.
Content enrichment	Enriches message content from an enrichment source. For example, if the original message contains an `AccountID`, the function can query a database using the `AccountID` to retrieve the customer region, add the region to the message body, and then send the message to the sink.
Content mapping	Maps message content based on a regular expression. For example, you can mask sensitive fields in messages or minimize message sizes.

This topic uses Message Queue for Apache Kafka to demonstrate how to use data cleansing.

Use cases

Content splitting

For example, a message contains a list of students.

message:
[John, Male, Class 4|Alice, Female, Class 3|David, Male, Class 4]

You can split the message into individual records for each student and then send these records to separate sinks, as shown below.

message:
    [John, Male, Class 4]
message:
    [Alice, Female, Class 3]
message:
    [David, Male, Class 4]

Dynamic routing

For example, a message contains a list of toothpaste products.

message:
[BrandA, toothpaste, $12.98, 100g
 BrandB, toothpaste, $7.99, 80g
 BrandC, toothpaste, $1.99, 100g]

You can route the products to different topics based on custom rules. The rules are as follows:

If a message starts with BrandA, send it to the BrandA-item-topic and BrandA-discount-topic.
If a message starts with BrandB, send it to the BrandB-item-topic and BrandB-discount-topic.
Other messages are sent to the Unknown-brand-topic.

The following JSON object describes these rules.

{
  "defaultTopic": "Unknown-brand-topic",
  "rules": [
    {
      "regex": "^BrandA",
      "targetTopics": [
        "BrandA-item-topic",
        "BrandA-discount-topic"
      ]
    },
    {
      "regex": "^BrandB",
      "targetTopics": [
        "BrandB-item-topic",
        "BrandB-discount-topic"
      ]
    }
  ]
}

Content enrichment

This example shows how to enrich a message that contains an IP address. Assume that a service access log is in the following format:

{
  "accountID": "164901546557****",
  "hostIP": "192.168.XX.XX"
}

To identify the IP address source, use a mapping stored in a MySQL database.

CREATE TABLE `tb_ip` (
              
The processed message is shown below.
    ->      `IP` VARCHAR(256) NOT NULL,
    ->     `Region` VARCHAR(256) NOT NULL,
    ->      `ISP` VARCHAR(256) NOT NULL,
    ->      PRIMARY KEY (`IP`)
    -> );

{
  "accountID": "164901546557****",
  "hostIP": "192.168.XX.XX",
  "region": "beijing"
}

Content mapping

For example, a message contains employee registration information that includes private data, such as employee IDs and phone numbers.

John, Employee ID 1, 131 1111 1111
              
You can mask private employee information before the message is sent to a sink, as shown below.
Alice, Employee ID 2, 132 2222 2222
David, Employee ID 3, 133 3333 3333

J***, Employee ID *, *** **** ****
A****, Employee ID *, *** **** ****
D****, Employee ID *, *** **** ****

Procedure

Log on to the ApsaraMQ for Kafka console. In the left-side navigation pane, choose Connector Ecosystem Integration > Tasks, and then click Create Task.

On the Create Task page, configure the event source, filtering rules, data cleansing template, and event sink in the Source, Filtering, Transformation, and Sink steps.

In the Source section, select China (Hangzhou) for Region, select the target Kafka instance, and select kafka_test for Topic.

① Source and ④ Sink

Select different instances of ApsaraMQ for Kafka.

② Filtering

This parameter is optional. If you do not set this parameter, all events are processed. For more information about matching rules, see event patterns.

③ Transformation

Select a template provided by Function Compute, such as Content Splitting, Content Mapping, Content Enrichment, or Dynamic Routing. You can select a template based on your business requirements. These templates provide basic data processing logic that you can use as-is or customize.

This example uses the Content Splitting template.

From the Alibaba Cloud Service drop-down list, select Function Compute (acs:fc:function). Select Create Function Template and choose Content Splitting (transform_split) from the Function Template list. In the generated function code, modify the delimiter variable based on your data format. The default value is |.