如何在Stream Analytics查询中拼接IoT Hub与Blob匹配的所有结果？

我明白你的需求——你想把IoT Hub输入和Blob Storage输入按ID匹配后，把所有匹配的Blob数据打包成一个数组输出，而不是每条匹配结果单独生成一条消息。既然Stream Analytics不支持GROUP_CONCAT，这里有几个实用的替代方案，按推荐优先级排序：

1. 使用内置的Collect函数（最直接方案） #

Stream Analytics其实提供了Collect函数，专门用来将分组内的多行记录聚合为一个JSON数组，完美替代GROUP_CONCAT的场景。你只需要结合GROUP BY和时间窗口（流处理必须指定窗口，因为数据是持续流入的）来实现：

SELECT 
    IoTHub.id AS DeviceId,
    Collect(Blob) AS MatchedBlobs
FROM iothub IoTHub
JOIN blob Blob ON IoTHub.id = Blob.deviceId
GROUP BY IoTHub.id, TumblingWindow(second, 10) -- 可根据业务调整窗口类型和大小
INTO [YourTargetOutput]

关键注意事项： #

• 窗口选择：这里用了翻滚窗口（TumblingWindow），你可以根据实际场景换成滑动窗口（SlidingWindow）或会话窗口（SessionWindow）。比如如果你的Blob数据是静态的、IoT消息是周期性的，可适当调大窗口大小避免重复聚合；如果Blob数据会更新，要确保窗口能覆盖到最新的Blob数据。
• 数组格式：Collect(Blob)会直接将匹配的Blob对象打包成数组，输出格式就是你想要的结构：

  {
    "DeviceId": "test001",
    "MatchedBlobs": [
      { "deviceId": "test001", "data": "Sample1" },
      { "deviceId": "test001", "data": "SampleX" }
    ]
  }
  

• 去重需求：如果同一个DeviceId存在重复的Blob记录，你可以在JOIN前先对Blob数据源做去重处理，比如：

  WITH DeduplicatedBlobs AS (
      SELECT DISTINCT * FROM blob
  )
  SELECT 
      IoTHub.id AS DeviceId,
      Collect(Blob) AS MatchedBlobs
  FROM iothub IoTHub
  JOIN DeduplicatedBlobs Blob ON IoTHub.id = Blob.deviceId
  GROUP BY IoTHub.id, TumblingWindow(second, 10)
  INTO [YourTargetOutput]
  

2. 自定义聚合函数（UDA）处理复杂逻辑 #

如果Collect函数无法满足你的特殊需求（比如需要对数组内的元素做自定义转换、过滤），可以创建自定义聚合函数（UDA）来实现更灵活的聚合：

步骤1：编写C# UDA代码 #

using System;
using System.Collections.Generic;
using Microsoft.Azure.StreamAnalytics;

[DataContract]
public class BlobArrayAggregator
{
    [DataMember]
    public List<BlobItem> BlobList { get; set; }

    public BlobArrayAggregator()
    {
        BlobList = new List<BlobItem>();
    }
}

[DataContract]
public class BlobItem
{
    [DataMember]
    public string deviceId { get; set; }
    [DataMember]
    public string data { get; set; }
}

public class CustomBlobArrayAggregate : AggregateFunction<BlobItem, BlobArrayAggregator>
{
    public override BlobArrayAggregator Init()
    {
        return new BlobArrayAggregator();
    }

    public override BlobArrayAggregator Accumulate(BlobArrayAggregator aggregator, BlobItem value)
    {
        // 这里可以添加自定义逻辑，比如去重、字段转换
        if (!aggregator.BlobList.Exists(b => b.deviceId == value.deviceId && b.data == value.data))
        {
            aggregator.BlobList.Add(value);
        }
        return aggregator;
    }

    public override BlobArrayAggregator ComputeResult(BlobArrayAggregator aggregator)
    {
        return aggregator;
    }
}

步骤2：在Stream Analytics中使用UDA #

将编译好的UDA上传到Stream Analytics作业，然后编写查询：

WITH JoinedData AS (
    SELECT IoTHub.id AS DeviceId, Blob
    FROM iothub IoTHub
    JOIN blob Blob ON IoTHub.id = Blob.deviceId
)
SELECT 
    DeviceId,
    CustomBlobArrayAggregate(Blob) AS MatchedBlobs
INTO [YourTargetOutput]
FROM JoinedData
GROUP BY DeviceId, TumblingWindow(second, 10)

3. 结合Azure Functions做后续聚合 #

如果不想在Stream Analytics中处理聚合，也可以先将所有匹配的原始结果输出到中间存储（比如Event Hub或Blob Storage），再用Azure Functions触发合并：

1. 第一步：运行原有JOIN查询，将所有匹配结果输出到Event Hub：

   SELECT Blob
   FROM iothub IoTHub
   JOIN blob Blob ON IoTHub.id = Blob.deviceId
   INTO [IntermediateEventHub]
   

2. 第二步：创建Azure Function，监听这个Event Hub，用缓存（如Redis）或持久化存储（如Table Storage）暂存同一个DeviceId的记录，当达到触发条件（比如窗口时间到期、收到预期数量的记录），将记录合并为数组后输出到最终目标。

这个方案适合需要更复杂的触发逻辑或外部依赖的场景，但需要额外的服务组件。

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