数据清洗规则_云消息队列 Kafka 版(Kafka)-阿里云帮助中心

数据清洗功能提供常见的消息处理模板，包括内容分割、动态路由、内容富化和内容映射等。您可以直接利用模板处理消息，也可以根据业务情况在模板基础上修改代码。

背景信息

消息数据清洗任务提供基本的算子能力，底层逻辑使用函数计算。支持进行数据清洗的产品包含云消息队列 RocketMQ 版、云消息队列 Kafka 版、云消息队列 MQTT 版、云消息队列 RabbitMQ 版轻量消息队列（原 MNS）。数据清洗任务创建完成后，您可以登录函数计算控制台，进行代码自定义及相应函数配置的修改。

算子	算子能力说明
内容分割	根据正则表达式对消息内容进行分割，将分割后的消息逐条发送至目标。
动态路由	根据正则表达式匹配消息内容，将匹配成功的消息路由至对应目标，将匹配不成功的消息路由至默认目标。
内容富化	根据富化源对消息内容进行富化。如果消息原始内容包含AccountID，处理时根据AccountID查询数据库，获得客户地域后填至源消息体中，并发送至目标服务。
内容映射	根据正则表达式对消息内容进行映射处理。例如，屏蔽消息中敏感字段或将消息大小缩减至最小标准。

本文以云消息队列 Kafka 版为例介绍如何使用数据清洗。

场景示例

内容分割

例如，以下是一份学生名单。

message:
[张三，男，4班|李四，女，3班|王五，男，4班]

需要将消息拆分为单个学生的信息，然后分三条消息推送至各目标服务。如下所示。

message:
    [张三，男，4班]
message:
    [李四，女，3班]
message:
    [王五，男，4班]

实现效果如下图。

动态路由

例如，以下是一份牙膏信息清单。

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

需要按照自定义动态规则，将列表路由至目标Topic。规则描述如下所示。

如果消息以BrandA开头，发送至BrandA-item-topic和BrandA-discount-topic这两个topic。
如果消息以BrandB开头，发送至BrandB-item-topic和BrandB-discount-topic这两个topic。
其余消息发送至Unknown-brand-topic。

规则的JSON描述如下。

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

实现效果如下图。

内容富化

本文以一个IP地址段处理的场景富化为例。假设某服务的访问日志如下所示。

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

需要统计IP地址的来源，并且映射关系存储于数据库MySQL。

CREATE TABLE `tb_ip` (
    ->      `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"
}

实现效果如下图。

内容映射

例如，以下是某公司员工登记信息，涉及了员工工号、电话号码等隐私内容。

张三，工号1，131 1111 1111
李四，工号2，132 2222 2222
王五，工号3，133 3333 3333

需要将以上消息中员工隐私信息进行屏蔽，然后推送至目标服务。如下所示。

张*，工号*，*** **** ****
李*，工号*，*** **** ****
王*，工号*，*** **** ****

实现效果如下图。

操作步骤

登录云消息队列 Kafka 版控制台，左侧导航栏选择Connector 生态集成 > 任务列表，然后单击创建任务。

在创建任务页面的Source（源）、Filtering（过滤）、Transform（转换）和Sink（目标）配置向导分别设置事件源、过滤规则、数据清洗模板和事件目标。

①Source（源）和④Sink（目标）

选择云消息队列 Kafka 版的不同实例。

②Filtering（过滤）

可选配置，如果不设置表示匹配全部事件，更多匹配规则，请参见消息过滤。

③Transform（转换）

选择一个函数计算提供的模板，包括内容分割、内容映射、内容富化和动态路由。您可以根据需求选择以上模板，模板中提供了基础的数据处理逻辑，可以直接使用也可以自行调整。

本文以选择内容分割模板为例。