ApsaraMQ for Kafka:Data cleansing

メッセージを個別のレコードに分割

複数のレコードを含む単一のメッセージを、レコードごとに 1 つの個別のメッセージに分割します。

入力：セミコロンで区切られた学生レコードのリストを含むメッセージ。

[Jack, Male, Class 4; Alice, Female, Class 3; John, Male, Class 4]

出力：それぞれが 1 つの学生レコードを含む 3 つの個別のメッセージ。

[アリス、女性、クラス 3]

[Alice, Female, Class 3]

[John, Male, Class 4]

コンテンツパターンによるメッセージのルーティング

コンテンツパターンに基づいて、メッセージを異なる送信先 Topic にルーティングします。

入力：複数のブランドの商品エントリを含むメッセージ。

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

ルーティングルール：

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

外部データによるメッセージのエンリッチメント

外部データソースでフィールド値を検索し、外部データをメッセージに追加します。

入力：アカウント ID とホスト 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`)
);

出力：region フィールドが追加された元のメッセージ。

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

メッセージ内の機密フィールドのマスク

正規表現を使用して、メッセージ内の特定のフィールドを変換またはマスクします。

入力：従業員 ID や電話番号などの機密データを含む従業員登録レコード。

James, Employee ID 1, 131 1111 1111
Mary, Employee ID 2, 132 2222 2222
David, Employee ID 3, 133 3333 3333

出力：機密フィールドがマスクされた同じレコード。

Ja*, Employee ID *, ***********
Ma*, Employee ID *, ***********
Dav*, Employee ID *, ***********