流分析聚合窗口

Posted 2023-05-08

【中文标题】流分析聚合窗口

【英文标题】：Stream Analytics Aggregation Window

【发布时间】：2018-12-13 04:46:55

【问题描述】：

我需要帮助\关于在扩展窗口上执行聚合时如何忽略旧事件的建议。我有流入事件中心的销售数据。 事件中心用作输入流。我需要产生两个指标 - 30 秒聚合（翻滚） - 全天累计销售价值，即从开门开始

门打开时间是可变的（动态的），因此我从 blob 中读取参考数据集；并将 Gateopen 日期时间加入销售流。 翻滚窗口上的 30 秒聚合工作正常。 给定门打开是可变的；我目前正在使用 12 小时跳跃窗口和 30 秒跳跃，并尝试通过使用 EventProcessDatetime > GateOpen 逻辑来限制要聚合的事件。

SELECT 
        Dateadd(ss,-30,System.Timestamp )  AS TimeSliceUTCStart
        , System.Timestamp AS TimeSliceUTCEnd   
        , p.Section                                       AS Section
        , SUM(CASE WHEN p.Classification = 'Retail' 
                AND p.ActivityDateTime > p.GateOpen THEN p.[sales_amt_gross] ELSE 0 END)    AS SaleTotalRetail


   FROM FilteredBase p 
   GROUP BY 
          p.Section
            , HoppingWindow(Duration(Hour, 12), hop(second, 30),Offset(millisecond, -1)) 

问题：我正在从前一天\timeslice 中汇总销售额。 总的来说，我想要达到的结果很简单。商店最多可营业 5、8、10 或 12 小时。随着时间的推移，我们希望能够像在直播中一样了解每个部分的销售情况。任何建议或提示将不胜感激。

【参考方案1】：

直观地说，查询看起来不错，但背后发生的事情是 Azure 流分析正在使用在每个时间窗口的时间有效的参考数据文件。然后，当它看到前一天的偶数时，它将使用当时存在的参考数据（这可能会使比较 p.ActivityDateTime > p.GateOpen 为前一天的开放时间为真）。

我按如下方式修改了查询（假设您每个部分每天有 1 个公开活动）。请让我知道这对你有没有用。如果没有，您能否发送一些示例数据，以便我可以相应地修改查询。我们将研究如何使这些查询更易于编写。

WITH thirdtysecReporting AS
(
    SELECT
        p.Section Section,
        DATETIMEFROMPARTS(DATEPART(year, System.Timestamp), DATEPART(month, System.Timestamp), DATEPART(day, System.Timestamp), 0, 0, 0, 0) as date,
        System.Timestamp Windowend,
        SUM(p.sales_amt_gross) thirtysecSales
    FROM input TIMESTAMP BY p.ActivityDateTime
    GROUP BY TumblingWindow(second, 30), p.Section
)

,hopping AS
(
    SELECT
        Section,
        System.Timestamp HopEnd,
        date,
        SUM(thirtysecSales) SumSales
    FROM thirdtysecReporting
    GROUP BY HoppingWindow(second, 86400, 30), Section, date -- Hopping on 24 hours, reported every 30 second
)

,filtered as -- This step ignores data from the previous day
(
    SELECT 
        Section,
        HopEnd,
        date,
        SUMQt = CASE
            WHEN DAY(HopEnd) = DAY(date) OR DATEPART(hour, HopEnd) = DATEPART(hour, date) THEN SumSales
            ELSE 0
        END
    FROM hopping
)

SELECT Section, -- Final query
        HopEnd,
        MAX(SUMQt) AS SumQt
FROM filtered
GROUP BY TumblingWindow(hour, 1), Section, hopend

谢谢，

JS - Azure 流分析