如何在Azure Stream Analytics中识别多数据集的公共数据

1. 前置元数据对齐 #

• 首先给所有写入Event Hub的数据源追加固定的数据源标识字段，比如源1标记source_tag = 'source_1'、源2标记source_tag = 'source_2'，该操作可在数据入库前完成，也可在Stream Analytics查询入口临时追加。
• 梳理不同数据源的同义字段映射关系，避免同一个业务字段在不同源命名不同导致匹配失效，比如源1的PNR No、源2的pnr_number统一映射为别名common_pnr，方便后续关联。
• 提前对公共字段做格式标准化，比如用TRIM()去掉首尾空格、UPPER()统一大小写，避免因格式差异导致匹配遗漏。

2. 基于Stream Analytics查询识别公共数据 #

如果需要找出同时出现在多个数据源中的公共数据行，可直接用流JOIN逻辑实现，示例查询代码如下：

WITH
-- 拆分源1数据并统一公共字段别名
source1 AS (
    SELECT
        [PNR No] AS common_pnr,
        other_field_a,
        source_tag
    FROM YourEventHubInput
    WHERE source_tag = 'source_1'
),
-- 拆分源2数据并统一公共字段别名
source2 AS (
    SELECT
        [PNR No] AS common_pnr,
        other_field_b,
        source_tag
    FROM YourEventHubInput
    WHERE source_tag = 'source_2'
)
-- 关联两个源得到公共PNR对应的数据
SELECT
    s1.common_pnr AS pnr_no,
    s1.other_field_a AS source1_field,
    s2.other_field_b AS source2_field
FROM source1 s1
JOIN source2 s2
ON s1.common_pnr = s2.common_pnr
-- 流数据JOIN必须指定时间窗口，可根据业务允许的最大数据延迟调整范围
WHERE DATEDIFF(second, s1, s2) BETWEEN 0 AND 3600

• 如果需要识别2个以上数据源的公共数据，叠加JOIN逻辑即可，每一次关联都需要按要求指定时间窗口范围。
• 如果是要统计多个数据集的公共字段列表，可先把所有数据源的字段名导出到Azure存储等介质，再聚合统计出现次数≥数据源数量的字段，即为所有数据集共有的公共字段。

3. 注意事项 #

• 时间窗口大小需要结合业务场景的数据到达延迟设置，窗口过小会导致因数据到达时间差漏匹配，窗口过大则会增大计算资源消耗。
• 高并发场景下如果公共字段重复值较多，可提前对数据按公共字段做分区，提升关联计算效率。

内容的提问来源于stack exchange，提问作者jsrathnayake