python - BigQuery - 在分区内嵌套操作，以便聚合具有 170 亿条记录的表中的连续记录

Question

我对 SQL 和 BigQuery 还很陌生，一周来我一直在努力寻找解决这个问题的可行解决方案。我拥有的两个解决方案无法扩展。

背景

拥有一个包含 170 亿条记录的 BigQuery 表。每条记录代表一次设备 ping。每条记录都包含时间戳、用于识别用户的 ID 以及接收 ping 的位置的名称。

获取此数据表，按 ID 对其进行分区并按时间戳进行排序。然后您将获得一组按时间顺序排列的 ping。用户可能对位置 A 执行 1 次 ping 操作，然后对位置 B 执行 7 次 ping 操作，然后对位置 C 执行 2 次 ping 操作，最后对 A 执行 2 次 ping 操作。

ID        timestamp             Location
ABC123    2017-10-12 10:20:37   A
ABC123    2017-10-12 11:15:21   B
ABC123    2017-10-12 11:21:47   B
ABC123    2017-10-12 11:25:05   B
ABC123    2017-10-12 11:32:12   B
ABC123    2017-10-12 11:36:24   B
ABC123    2017-10-12 11:47:13   B
ABC123    2017-10-12 11:59:08   B
ABC123    2017-10-12 12:04:42   C
ABC123    2017-10-12 17:04:52   C
ABC123    2017-10-12 19:15:37   A
ABC123    2017-10-12 19:18:37   A

我想做的是，使用这张表并生成一个新表，每个“行程”一行。其中行程是一组连续的 ping，具有“first_ping”和“last_ping”列。如果一次行程包含 1 个 ping，则该时间戳既是第一个 ping，也是最后一个 ping。

ID        first_ping            last_ping             Location
ABC123    2017-10-12 10:20:37   2017-10-12 10:20:37   A
ABC123    2017-10-12 11:15:21   2017-10-12 11:59:08   B
ABC123    2017-10-12 12:04:42   2017-10-12 17:04:52   C
ABC123    2017-10-12 19:15:37   2017-10-12 19:18:37   A

尝试解决方案

Python

我从来没有处理过这么大的数据，而且我一直使用 Python。所以我第一次尝试的解决方案是一个 Python 脚本:

1. 查询BQ某个ID的所有数据
2. 按时间戳对数据进行排序
3. 使用“位置”上的 diff 函数来确定其何时发生变化
4. 使用 cumsum() 标记每组 ping 的所有项目具有相同的值。
5. 在 cumsum() 上使用 df.groupby() 获取每条记录一行，并使用first() 和 last() 获取first_ping 和last_ping 值。

此解决方案可生成我需要的输出，但对于 170 亿条记录和 69M 个唯一 ID 来说不可行。每个 ID 大约需要 10 秒，即大约 19 万小时的运行时间。

SQL

WITH visitWithIsChange AS 
(select
   *,
   LAG(location,1,'') OVER (PARTITION BY user_id ORDER BY timestamp) previous,
    CASE 
     WHEN (LAG(location,1,'') 
           OVER (PARTITION BY user_id ORDER BY timestamp)) = location
           THEN 0
           ELSE 1
     END ischange
 FROM `ping_table` ORDER BY user_id, timestamp),
 visitsWithcumsum AS (
   SELECT 
      t1.*,
      SUM(t2.ischange) AS cumulativeSum 
   FROM visitWithIsChange t1
        INNER JOIN
             visitWithIsChange t2
               ON
                 t1.local_timestamp >=t2.local_timestamp
                 AND
                 t1.user_id=t2.user_id
   GROUP BY 
     t1.local_timestamp,
     t1.user_id,
     t1.chain_id,
     t1.previous,
     t1.isChange
   ORDER BY user_id, timestamp
)
SELECT 
  MIN(timestamp) AS first_ping,
  MAX(local_timestamp) AS last_ping,
  user_id,
  chain_id,
FROM visitsWithcumsum
GROUP BY
  user_id,
  cumulativeSum,
  chain_id,
ORDER BY user_id, first_ping

我知道SQL语句的问题是在分区之外使用了ORDER BY。每次对超过几十万行调用 ORDER BY 时，BigQuery 都会抛出资源超出错误。我的理解是，发生这种情况是因为 ORDER BY 语句要求所有被排序的数据都驻留在一个节点上。

问题

1. 是否可以使用上述 SQL 方法，但将所有逻辑嵌套在分区语句中？基本上，按 user_id 分区并在该分区内执行所有排序、累积和等？
2. 有更好的方法来解决这个问题吗？

我感谢所有的意见。我完全不知道解决这个问题的最佳方法，并且感觉超出了我的能力范围。

Answer 1

应该使用累积和而不是非等值连接来计算cumulativeSum:

WITH visitWithIsChange AS 
(select
   *,
    CASE 
     WHEN (LAG(location,1,'') 
           OVER (PARTITION BY user_id ORDER BY timestamp)) = location
           THEN 0
           ELSE 1
     END ischange
 FROM `ping_table`
 -- I don't now about BigQuery, but why do you need this?
 --ORDER BY user_id, timestamp
 ),
 visitsWithcumsum AS (
   SELECT 
      *,
      SUM(ischange)
      OVER (PARTITION BY user_id
            ORDER BY timestamp
            ROWS UNBOUNDED PREDECING) AS cumulativeSum 
   FROM visitWithIsChange  
)
SELECT 
  MIN(timestamp) AS first_ping,
  MAX(local_timestamp) AS last_ping,
  user_id,
  chain_id,
FROM visitsWithcumsum
GROUP BY
  user_id,
  cumulativeSum,
  chain_id,
ORDER BY user_id, first_ping