Stream Analytics查询：如何实现同输入数据源的自连接？

首先，我们来拆解你遇到的几个核心问题，一步步帮你理清逻辑并解决问题：

1. 返回值为0的真实原因 #

你看到的value字段为0，并不是连接失败导致的，而是整数除法的结果！你的查询里用了cast(I1.Value as bigint)/30，bigint是整数类型，Stream Analytics执行整数除法时会直接舍去小数部分。比如你的I1.Value是10，10除以30的整数结果就是0。

要得到正确的小数结果，只需把其中一个操作数转为浮点类型即可：

cast(I1.Value as float)/30 AS value

2. DATEDIFF(MINUTE,I1,I2) BETWEEN 0 AND 1的作用 #

这个条件有两个关键作用：

• 匹配时间相近的记录：你的两个标签EngineFuelConsumption和ShaftsRunning的时间戳大概率不会完全一致（比如你提供的示例数据里，一个是12:34:34，另一个是12:35:34，差1分钟），这个条件用来匹配时间差在0到1分钟内的成对记录。
• 满足流处理的状态约束：和传统SQL不同，Stream Analytics作为流处理引擎，要求流之间的JOIN必须包含时间窗口约束——因为数据流是无限的，系统无法永久保存所有历史数据来匹配后续记录。这个条件告诉系统只需要保留最近1分钟内的记录用于匹配，避免内存溢出。

3. 移除该条件报错的原因 #

正如上面所说，Stream Analytics必须通过时间范围约束来控制JOIN操作的状态生命周期。如果没有这个约束，查询会被视为“无边界”的JOIN，系统无法确定需要保存多少历史数据，因此会直接报错拒绝执行。这是流处理SQL和传统SQL的核心区别之一。

4. 修复你的连接逻辑 #

你当前的查询同时用了I1.Time=I2.Time和DATEDIFF(...)，这两个条件是冲突的：示例数据里两个时间戳差1分钟，I1.Time=I2.Time完全不满足，导致JOIN无法匹配到记录（你说返回了行，可能是实际数据中有部分时间戳一致的记录，但因为整数除法得到0）。

正确的做法是去掉严格的I1.Time=I2.Time条件，只保留时间范围匹配：

WITH DataInput1 AS (
    SELECT DATA.Fqn AS fqn, DATA.Value AS value, DATA.time AS time
    FROM (
        SELECT Tag.ArrayValue.Fqn AS fqn, VQT.ArrayValue.V AS value, VQT.ArrayValue.T AS time
        FROM MetsoQuakeFuelCon AS TimeSeries
        CROSS APPLY GetArrayElements(TimeSeries.[timeSeries]) AS Tag
        CROSS APPLY GetArrayElements(Tag.ArrayValue.vqts) AS VQT
    ) AS DATA
    WHERE DATA.fqn like '%EngineFuelConsumption'
),
DataInput2 AS (
    SELECT DATA.Fqn AS fqn, DATA.Value AS value, DATA.time AS time
    FROM (
        SELECT Tag.ArrayValue.Fqn AS fqn, VQT.ArrayValue.V AS value, VQT.ArrayValue.T AS time
        FROM MetsoQuakeFuelCon AS TimeSeries
        CROSS APPLY GetArrayElements(TimeSeries.[timeSeries]) AS Tag
        CROSS APPLY GetArrayElements(Tag.ArrayValue.vqts) AS VQT
    ) AS DATA
    WHERE DATA.fqn like '%ShaftsRunning' and DATA.Value like '1'
),
DataInput as (
    select 
        I1.Fqn AS fqn, 
        cast(I1.Value as float)/30 AS value,  -- 改为浮点除法得到正确小数结果
        DATETIMEFROMPARTS(DATEPART(year,I1.Time ),DATEPART(month,I1.Time ),DATEPART(day,I1.Time ) ,DATEPART(hour,I1.Time ),00,00,00 ) AS time
    from DataInput1 I1
    JOIN DataInput2 I2 
        ON DATEDIFF(MINUTE,I1.Time, I2.Time) BETWEEN 0 AND 1  -- 只保留时间范围匹配
)
select * from DataInput

如果需要更精确的时间窗口控制，你也可以尝试Stream Analytics的SESSION WINDOW或HOPPING WINDOW函数，但对于你的场景，调整后的JOIN条件已经足够解决问题。

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