BigQuery - 股票分类帐的窗口化

Question

我（认为）我有一个有趣的窗口场景，与网站库存可用性跟踪相关，可以根据标准 SQL 进行处理。这是试图根据从给定产品的“现有”数量的库存中推送和拉取的事件，随着时间的推移构建库存头寸视图

我们有 3 种与此问题相关的事件类型：

StocklevelUpdated（推送）：每晚午夜，我们都会从仓库中获取给定产品的 onHandQty 可用性水平的新更新。如果每个产品的 onHandQty 计为第二天交易的新值，这基本上是一个硬“重置”。 （注意：这实际上每晚都会发送一条记录，即使没有变化）。

OrderAccepted (PULL)：然后在一天中，有许多产品的“OrderAccepted”事件，在这种情况下，库存“onHandQtyDelta”的值为负（现在更少出售）。订购 2 件产品时为 -2。

OrderCancelled：产品的数量也可以取消，这对“onHandQtyDelta”具有正值，因为它会重新添加到可供出售的库存中）

下面是一个稍微简化版本的数据的表格视图，按我想要实现的时间顺序（注意：这显示了一个产品，但当然有很多）。 p>

onHandQtyDelta - 由于此事件而对 onHandQty 的更改

onHandQty - 这是增量影响发布那个时间点的净正面。

现在虽然上图显示了所有值都很好地显示出来（尽管注释 29 是这些午夜重置之一），但实际上，并非所有这些数据都可用并且其中一个值需要为这些订单事件类型中的每一个派生。即缺少 1。

onHandQty：实际数据集中唯一定义了绝对 onHandQty 的行是“StocklevelUpdated”事件。本质上，这会在午夜“重置”产品的此值（例如 29）。日志需要基本上追溯到最近的这些。但是，需要导出 onHandQtyDelta。

onHandQtyDelta 只有 OrderAccepts 和 OrderCancelled 事件有此值，需要用于计算 onHandQty。

所以一张图片说一千个单词，所以要处理的数据的实际情况如下所示：

如何有效地执行此操作（假设有数百万行）这样做？

我的想法是使用windowing和'lag'函数来回顾previous记录的onHandQty值，看看它是什么，然后做加法或减法得出新的onHandQty价值。

问题是递归问题，因为前一个事件本身需要回顾它的前一个，依此类推....直到您到达 stocklevelUpdated 事件，因为这是唯一具有实际值的事件工作从。但是，当您不知道要返回多远才能获得此类事件时，如何使用窗口来执行此操作 - 可以是任意数量的 OrderAccepts 和 Cancells （或没有！）

也许对数组有些聪明，将给定的产品行收集到一个数组中并执行一些数组聚合函数？

我认为我一直认为窗口是一种方法，并且可能被一个简单的解决方案所困扰！很抱歉提供了所有细节，但不想对我需要帮助的内容含糊其辞。

下面给出了要处理的起始测试数据集（我只是按产品和时间对其进行排序，以便创建图像）

WITH stock_changes AS (
SELECT
  "StocklevelUpdated" AS eventName, 
  Timestamp("2017-06-29T23:59:59") AS stockLevelEventAt,
  "PRODUCT_190035001612" AS productId,
  null AS onHandQtyDelta,
  23 AS onHandQty
UNION ALL (
SELECT
  "StocklevelUpdated" AS eventName, 
  Timestamp("2017-06-29T23:59:59") AS stockLevelEventAt,
  "PRODUCT_4545423454545" AS productId,
  null AS onHandQtyDelta,
  120 AS onHandQty)
UNION ALL (
  SELECT 
  "OrderAccepted" AS eventName, 
  Timestamp("2017-06-30T01:02:20") AS stockLevelEventAt,
  "PRODUCT_190035001612" AS productId,
  -2 AS onHandQtyDelta,
  null AS onHandQty)
UNION ALL (
  SELECT 
  "OrderAccepted" AS eventName, 
  Timestamp("2017-06-30T02:19:20") AS stockLevelEventAt,
  "PRODUCT_190035001612" AS productId,
  -3 AS onHandQtyDelta,
  null AS onHandQty)
UNION ALL (
  SELECT 
  "OrderAccepted" AS eventName, 
  Timestamp("2017-06-30T05:13:20") AS stockLevelEventAt,
  "PRODUCT_4545423454545" AS productId,
  -3 AS onHandQtyDelta,
  null AS onHandQty)
UNION ALL (
  SELECT 
  "OrderCancelled" AS eventName, 
  Timestamp("2017-06-30T13:02:20") AS stockLevelEventAt,
  "PRODUCT_190035001612" AS productId,
  +2 AS onHandQtyDelta,
  null AS onHandQty)
UNION ALL (
  SELECT 
  "OrderCancelled" AS eventName, 
  Timestamp("2017-06-30T11:02:20") AS stockLevelEventAt,
  "PRODUCT_4545423454545" AS productId,
  2 AS onHandQtyDelta,
  null AS onHandQty)
UNION ALL (
SELECT
  "StocklevelUpdated" AS eventName, 
  Timestamp("2017-06-30T23:59:59") AS stockLevelEventAt,
  "PRODUCT_190035001612" AS productId,
  null AS onHandQtyDelta,
  29 AS onHandQty)
UNION ALL (
SELECT
  "StocklevelUpdated" AS eventName, 
  Timestamp("2017-06-30T23:59:59") AS stockLevelEventAt,
  "PRODUCT_4545423454545" AS productId,
  null AS onHandQtyDelta,
  140 AS onHandQty)
)
SELECT *
FROM stock_changes
order by productId, stockLevelEventAt ASC

Answer 1

以下是 BigQuery 标准 SQL

#standardSQL
WITH stock_changes AS (
  SELECT "StocklevelUpdated" AS eventName, TIMESTAMP("2017-06-29T23:59:59") AS stockLevelEventAt,
    "PRODUCT_190035001612" AS productId, NULL AS onHandQtyDelta, 23 AS onHandQty UNION ALL 
  SELECT "StocklevelUpdated", TIMESTAMP("2017-06-29T23:59:59"),"PRODUCT_4545423454545",NULL, 120 UNION ALL 
  SELECT "OrderAccepted", TIMESTAMP("2017-06-30T01:02:20"),"PRODUCT_190035001612",-2, NULL UNION ALL 
  SELECT "OrderAccepted", TIMESTAMP("2017-06-30T02:19:20"),"PRODUCT_190035001612",-3, NULL UNION ALL 
  SELECT "OrderAccepted", TIMESTAMP("2017-06-30T05:13:20"),"PRODUCT_4545423454545",-3, NULL UNION ALL 
  SELECT "OrderCancelled", TIMESTAMP("2017-06-30T13:02:20"),"PRODUCT_190035001612",+2, NULL UNION ALL 
  SELECT "OrderCancelled", TIMESTAMP("2017-06-30T11:02:20"),"PRODUCT_4545423454545",2, NULL UNION ALL 
  SELECT "StocklevelUpdated", TIMESTAMP("2017-06-30T23:59:59"),"PRODUCT_190035001612",NULL, 29 UNION ALL 
  SELECT "StocklevelUpdated", TIMESTAMP("2017-06-30T23:59:59"),"PRODUCT_4545423454545",NULL, 140
)
SELECT
  eventName, stockLevelEventAt, productId, 
  delta AS onHandQtyDelta, IFNULL(onHandQty, onHand ) AS onHandQty
FROM (
  SELECT *,
    SUM(IFNULL(onHandQty,0) - delta) 
      OVER(PARTITION BY productId, format_timestamp('%Y-%m-%d', stockLevelEventAt) 
      ORDER BY stockLevelEventAt DESC 
      rows BETWEEN UNBOUNDED PRECEDING AND 1 PRECEDING ) AS onHand
  FROM (
    SELECT eventName, stockLevelEventAt, productId, onHandQty, 
      CASE 
        WHEN prev IS NULL THEN IFNULL(onHandQtyDelta, 0) 
        ELSE onHandQty - prev - delta 
      END AS delta
    FROM (
      SELECT *, 
        SUM(IFNULL(onHandQtyDelta,0)) OVER(PARTITION BY productId, format_timestamp('%Y-%m-%d', stockLevelEventAt) ORDER BY stockLevelEventAt) AS delta,
        LAG(onHandQty) OVER(PARTITION BY productId, eventName ORDER BY stockLevelEventAt) AS prev
      FROM stock_changes
    )
  )
)
ORDER BY productId, stockLevelEventAt ASC   

结果如下

Row eventName           stockLevelEventAt       productId    onHandQtyDelta onHandQty    
1   StocklevelUpdated   2017-06-29 23:59:59 UTC PRODUCT_190035001612    0   23   
2   OrderAccepted       2017-06-30 01:02:20 UTC PRODUCT_190035001612    -2  21   
3   OrderAccepted       2017-06-30 02:19:20 UTC PRODUCT_190035001612    -3  18   
4   OrderCancelled      2017-06-30 13:02:20 UTC PRODUCT_190035001612    2   20   
5   StocklevelUpdated   2017-06-30 23:59:59 UTC PRODUCT_190035001612    9   29   
6   StocklevelUpdated   2017-06-29 23:59:59 UTC PRODUCT_4545423454545   0   120  
7   OrderAccepted       2017-06-30 05:13:20 UTC PRODUCT_4545423454545   -3  117  
8   OrderCancelled      2017-06-30 11:02:20 UTC PRODUCT_4545423454545   2   119  
9   StocklevelUpdated   2017-06-30 23:59:59 UTC PRODUCT_4545423454545   21  140  

很可能可以进一步优化 - 但我更专注于实现逻辑而不是优化