使用函数计算实现消息数据清洗

消息数据清洗功能提供常见的消息处理模板,支持对不同来源、不同格式的数据进行分割、动态路由和消息富化等。您可以直接利用函数计算提供的模板处理消息,也可以根据业务情况在模板基础上修改代码。本文以云消息队列 RocketMQ 版数据清洗为例,介绍消息处理模板的类型和使用方式。

功能介绍

消息数据清洗任务提供基本的算子能力,底层逻辑使用函数计算。云消息队列 RocketMQ 版消息数据清洗任务创建完成后,您可以登录函数计算,进行代码自定义及相应函数配置的修改。

算子

算子能力说明

消息过滤

按照正则表达式匹配消息内容,将匹配成功的消息发送至目标。更多信息,请参见事件模式

消息转换

根据字符串匹配,进行消息内容替换,例如字符大小写转换。将转换后的消息发送至目标。更多信息,请参见事件内容转换

内容分割

根据正则表达式对消息内容进行分割,将分割后的消息逐条发送至目标。

动态路由

根据正则表达式匹配消息内容,将匹配成功的消息路由至对应目标,将匹配不成功的消息路由至默认目标。

内容富化

根据富化源对消息内容进行富化。例如,消息原始内容包含AccountID,处理时根据AccountID查询数据库,获得客户地域后填至源消息体中,并发送至目标服务。

内容映射

根据正则表达式对消息内容进行映射处理。例如,屏蔽消息中敏感字段或将消息大小缩减至最小标准。

场景示例

内容分割

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

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

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

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

实现效果如下图。

image

动态路由

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

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"
      ]
    }
  ]
}

实现效果如下图。

image

内容富化

本文以一个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"
}

实现效果如下图。

image

内容映射

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

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

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

张*,工号*,*** **** ****
李*,工号*,*** **** ****
王*,工号*,*** **** ****

实现效果如下图。

image

操作步骤

1.创建数据清洗任务

  1. 登录事件总线 EventBridge控制台,左侧导航栏,选择事件流,然后单击创建事件流

  2. 创建事件流页面的Source(源)Filtering(过滤)Transform(转换)Sink(目标)配置向导分别设置事件源、过滤规则、数据清洗模板和事件目标。

    image

    Source(源)和④Sink(目标)

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

    Filtering(过滤)

    可选配置,如果不设置表示匹配全部事件,更多匹配规则,请参见事件模式

    Transform(转换)

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

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

    image

2.测试结果

  1. 登录消息队列 RocketMQ 版控制台,找到上一步配置的Source(源)实例,在目标topic右侧操作列,单击更多 > 快速体验,然后在快速体验的消息生产和消费面板,输入原始消息,单击确定即可发送消息。

    image

  2. 消息队列 RocketMQ 版控制台,找到上一步配置的Sink(目标)实例,依次单击消息查询 > 查询查询消息,您可以看到上一步发送的消息发送到目标端时已被分割为3条消息,依次单击消息行的详情,可以看到3条消息分别为 "data": "张三,男,4班" "data": "李四,女,3班""data": "王五,男,4班"

    image