Azure Stream Analytics作业触发Event Hub接收器超限错误如何排查？

问题原因分析 #

这个报错的核心是Event Hub单分区单消费者组最多支持5个并发接收器，你的作业配置虽然逻辑上符合WITH复用输入的规范，但触发了Azure Stream Analytics（ASA）的隐式多接收器读取逻辑，具体原因如下：

1. UNION双分支隐式重复读取输入：你当前的CombineAnalogAndDigital CTE中UNION的两个分支都独立引用了ODSMeasurements，ASA查询优化器处理CROSS APPLY + 参考表JOIN的组合逻辑时，不会自动缓存输入读取结果，会为两个分支分别创建Event Hub接收器，直接占用2个配额。
2. 高并行度放大接收器数量：如果你的作业配置了≥2个流单元（SU），ASA会自动为每个逻辑读取分支创建多个并行接收器，2个逻辑分支 × 2个SU就会产生4个接收器，再加上作业重启时未及时释放的旧连接，很容易超过5个的配额。
3. 消费者组残留连接：多次重启作业后，Event Hub端未超时释放的旧AMQP连接会占用接收器配额，你诊断日志中的5个nil记录就是已经断开但还未被回收的无效接收器。
4. 输出分支未复用计算结果：4个独立的INTO输出如果没有触发ASA的公共结果缓存，也可能额外增加读取次数。

解决办法 #

1. 重构查询避免重复读取输入 #

将UNION的两个分支合并为单分支，只读取一次Event Hub输入：

WITH 
ODSMeasurements AS (
    SELECT 
        collectionTimestamp,
        analogValues,
        digitalValues,
        type,
        translationTable
    FROM EventhubODSMeasurements
    TIMESTAMP BY CAST(CONCAT(SUBSTRING(collectionTimestamp, 1, 10), ' ', SUBSTRING(collectionTimestamp, 12, 12)) AS datetime) -- 提前定义时间戳避免重复计算
),
-- 一次性展开模拟量和数字量
AllMeasurements AS (
    SELECT
        collectionTimestamp,
        type AS Status,
        translationTable,
        'analog' AS valueType,
        PropertyName AS Tag,
        PropertyValue.value AS rawValue
    FROM ODSMeasurements
    CROSS APPLY GetRecordProperties(analogValues)
    
    UNION ALL -- 两类数据无重复直接用UNION ALL提升性能
    
    SELECT
        collectionTimestamp,
        type AS Status,
        translationTable,
        'digital' AS valueType,
        PropertyName AS Tag,
        PropertyValue.value AS rawValue
    FROM ODSMeasurements
    CROSS APPLY GetRecordProperties(digitalValues)
),
CombineAnalogAndDigital AS (
    SELECT 
        System.Timestamp AS "TimeStamp",
        CASE WHEN valueType = 'analog' THEN ROUND(rawValue / CAST(TT.ConversionFactor AS float),5) ELSE CAST(-9999.00000 AS float) END AS "ValueNumber",
        CASE WHEN valueType = 'digital' THEN CAST(rawValue AS nvarchar(max)) ELSE NULL END AS "ValueBit",
        CAST(TT.MeasurementTypeId AS bigint) AS "MeasurementTypeId",
        TT.MeasurementTypeName AS "MeasurementName",
        TT.PartName AS "PartName",
        CAST(TT.ElementId AS bigint) AS "ElementId",
        TT.ElementName AS "ElementName",
        TT.ObjectName AS "ObjectName",
        TT.LocationName AS "LocationName",
        CAST(TT.TranslationTableId AS bigint) AS "TranslationTableId",
        Status
    FROM AllMeasurements AM
    INNER JOIN SQLTranslationTable TT
    ON 
        TT.Tag = AM.Tag AND
        CAST(TT.Version as bigint) = AM.translationTable.version AND
        TT.Name = AM.translationTable.name
)

-- 后续输出逻辑保持不变
SELECT *
INTO DatalakeHarmonizedMeasurements
FROM CombineAnalogAndDigital
PARTITION BY TranslationTableId

SELECT *
INTO FunctionsHarmonizedMeasurements
FROM CombineAnalogAndDigital

SELECT Timestamp, ValueNumber, CAST(ValueBit AS bit) AS ValueBit, ElementId, MeasurementTypeId, CAST(TranslationTableId AS bigint) AS TranslationTableId
INTO SQLRealtimeMeasurements
FROM CombineAnalogAndDigital

SELECT *
INTO EventHubHarmonizedMeasurements
FROM CombineAnalogAndDigital
PARTITION BY TranslationTableId

2. 调整并行度配置 #

• 检查Event Hub分区数，将ASA作业的流单元数设置为≤Event Hub分区数，避免多个SU实例读取同一个Event Hub分区
• 开启输入分区对齐：在Event Hub输入配置中设置PartitionKey为PartitionId，让每个ASA并行实例只处理一个Event Hub分区

3. 清理残留连接 #

• 如果是刚重启作业出现的报错，等待1-2分钟让Event Hub自动回收超时的旧连接
• 确认streamanalytics消费者组没有被其他应用（函数、其他流作业、测试工具）占用，必要时新建一个独立消费者组替换现有配置

4. 开启查询优化 #

在ASA作业的「配置」-「查询优化」中开启对应选项，让ASA自动复用公共计算结果，避免多个输出分支重复读取输入。

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