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. 查询一个ID的所有数据的BQ

2. 按时间戳对数据进行排序

3. 使用“位置”上的 diff 函数来确定它何时发生变化

4. 使用 cumsum() 用相同的值标记每组 ping 的所有项目。

5. 在 cumsum() 上使用 df.groupby() 获取每条记录的一行，并使用 first() 和 last() 获取 first_ping 和 last_ping 值。

这个解决方案产生了我需要的输出，但对于 170 亿条记录和 6900 万个唯一 ID 是不可行的。每个 ID 大约需要 10 秒，即大约 190K 小时的运行时间。

问题

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 visitsWithcumsumGROUP BY  user_id,  cumulativeSum,  chain_id,ORDER BY user_id, first_ping