如何在Azure Stream Analytics中移除EventHub重复事件？

当然可以在Stream Analytics端直接处理重复数据，完全不需要修改已有的CosmosDB集合。针对你的场景，我推荐两种高效的去重方案，结合你的现有查询调整即可：

方案1：使用ROW_NUMBER()窗口函数精准去重（推荐） #

这种方法通过给重复的事件组分配序号，只保留每组的第一条数据，灵活性很高，可自定义判断“重复”的字段组合。

根据你提供的重复数据特征（idnum、basetime、time、sig1以及嵌套的sig2内容完全一致），可以修改你的查询，新增一个去重CTE：

WITH temp AS (
 SELECT 
  input.idnum AS IDNUM, 
  input.basetime AS BASETIME, 
  input.time AS TIME, 
  ROUND(input.sig1,5) AS SIG1, 
  flatArrayElement as SIG2, 
  udf.sgnlArrayMap(input.signals, input.basetime) AS SGNL,
  input.sig3TriggeredDateTime AS TRIGGER_TIME -- 加入触发时间作为去重标识的一部分
 FROM [input01] as input
 CROSS APPLY GetArrayElements(input.sig2) AS flatArrayElement
 WHERE GetArrayLength(input.sig2) >=1
),
SIGNALS AS (
 SELECT * FROM temp T JOIN master M ON T.SIG2.ArrayValue.sig3 = M.sig3
),
deduped_signals AS (
 SELECT *,
  ROW_NUMBER() OVER (
   PARTITION BY IDNUM, BASETIME, TIME, SIG1, SIG2.ArrayValue.id, SIG2.ArrayValue.sig3, TRIGGER_TIME
   ORDER BY (SELECT NULL) -- 若无需特定顺序，用此即可；若要保留最早/最晚事件，可改为 ORDER BY input.EventEnqueuedUtcTime
  ) AS rn
 FROM SIGNALS
)
--Insert SIG2 to COSMOS Container
SELECT t.IDNUM, t.BASETIME, t.TIME, t.SIG1, t.SIG2.ArrayValue.id AS ID, t.SIG2.ArrayValue.sig3 AS SIG3, t.SGNL 
INTO [CosmosTbl]
FROM deduped_signals t
WHERE t.rn = 1 -- 仅保留每组的第一条数据，去除重复
PARTITION BY PartitionId

关键说明： #

• PARTITION BY后面的字段是判断重复的依据，只要这些字段完全匹配，就会被视为同一组重复数据，你可以根据实际需求调整字段列表。
• 如果EventHub的生产者发送事件时设置了唯一的EventId，可以把input.EventId加入PARTITION BY，这样去重会更精准（避免误判内容相似但实际是不同的事件）。
• ORDER BY子句如果用input.EventEnqueuedUtcTime，可以选择保留最早或最晚到达的事件，更符合实际业务逻辑。

方案2：使用DISTINCT快速去重（适合完全重复的场景） #

如果你的重复事件是所有输出字段完全一致（包括嵌套数组和UDF处理后的SGNL），可以直接用DISTINCT关键字简化查询：

WITH temp AS (
 SELECT 
  input.idnum AS IDNUM, 
  input.basetime AS BASETIME, 
  input.time AS TIME, 
  ROUND(input.sig1,5) AS SIG1, 
  flatArrayElement as SIG2, 
  udf.sgnlArrayMap(input.signals, input.basetime) AS SGNL
 FROM [input01] as input
 CROSS APPLY GetArrayElements(input.sig2) AS flatArrayElement
 WHERE GetArrayLength(input.sig2) >=1
),
SIGNALS AS (
 SELECT * FROM temp T JOIN master M ON T.SIG2.ArrayValue.sig3 = M.sig3
)
--Insert SIG2 to COSMOS Container
SELECT DISTINCT t.IDNUM, t.BASETIME, t.TIME, t.SIG1, t.SIG2.ArrayValue.id AS ID, t.SIG2.ArrayValue.sig3 AS SIG3, t.SGNL 
INTO [CosmosTbl]
FROM SIGNALS t
PARTITION BY PartitionId

注意事项： #

• 这种方法仅适用于所有输出字段完全相同的重复项，对于部分字段相同的场景不适用。
• Azure Stream Analytics对复杂类型（如数组）的DISTINCT比较是支持的，但要确保UDF返回的SGNL在重复事件中完全一致。

额外优化建议 #

• 时间窗口限制：如果重复数据只会在特定时间范围内出现（比如5分钟内），可以结合窗口函数缩小去重范围，减少计算负载：

  deduped_signals AS (
   SELECT *,
    ROW_NUMBER() OVER (
     PARTITION BY IDNUM, BASETIME, TIME, SIG1, SIG2.ArrayValue.id, SIG2.ArrayValue.sig3
     ORDER BY input.EventEnqueuedUtcTime
     RANGE BETWEEN INTERVAL 5 MINUTE PRECEDING AND CURRENT ROW
    ) AS rn
   FROM SIGNALS
  )
  

• SU配置：去重操作会增加计算压力，确保你的Stream Analytics作业配置了足够的Streaming Units（SU），避免出现延迟。

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