Function Compute:Function Compute を使用した ApsaraMQ for RocketMQ メッセージデータのクレンジング

たとえば、元のメッセージである学生リスト [田中一郎, 男性, Class 4|佐藤花子, 女性, Class 3|山田健太, 男性, Class 4] を3つの個別のメッセージに分割し、それぞれを送信先サービスにプッシュする必要があります。これを行うには、コンテンツ分割オペレーターを使用します。分割されたメッセージは次のとおりです。

メッセージ:
    [張 三、男性、クラス 4]
メッセージ:
    [李 四、女性、クラス 3]
メッセージ:
    [王 五、男性、クラス 4]

次の図に結果を示します。

たとえば、以下は歯磨き粉情報のリストです。

メッセージ:
[BrandA, toothpaste, $12.98, 100g
 BrandB, toothpaste, $7.99, 80g
 BrandC, toothpaste, $1.99, 100g]

カスタム動的ルールに基づいて、リストを送信先トピックにルーティングする必要があります。ルールは次のように記述されます。

• メッセージがブランドAで始まる場合、BrandA-item-topic と BrandA-discount-topic に送信します。

• メッセージがブランドBで始まる場合、BrandB-item-topic と BrandB-discount-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"
}

次の図に結果を示します。

たとえば、以下は会社の従業員の登録情報であり、従業員IDや電話番号などの個人情報が含まれています。

Zhang San, Employee ID 1, 131 1111 1111
Li Si, Employee ID 2, 132 2222 2222
Wang Wu, Employee ID 3, 133 3333 3333

メッセージ内の従業員の個人情報をマスクし、送信先サービスにプッシュする必要があります。結果は次のとおりです。

Zhang*、従業員ID *、*** **** ****
Li*、従業員ID *、*** **** ****
Wang*、従業員ID *、*** **** ****

次の図に結果を示します。