ApsaraMQ for Kafka:データクレンジング

たとえば、メッセージに学生のリストが含まれているとします。

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

以下に示すように、メッセージを各学生の個別のレコードに分割し、これらのレコードを別々のシンクに送信できます。

message:
    [John, Male, Class 4]
message:
    [Alice, Female, Class 3]
message:
    [David, Male, Class 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 に送信します。

• メッセージが BrandB で始まる場合、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 や電話番号などの個人情報を含む従業員登録情報が含まれているとします。

以下に示すように、メッセージがシンクに送信される前に、従業員の個人情報をマスクできます。John, Employee ID 1, 131 1111 1111
Alice, Employee ID 2, 132 2222 2222
David, Employee ID 3, 133 3333 3333

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