如何用SQL（BigQuery）实现基于时间与区域的人员流动路径统计

需求说明 #

• 基于包含时间戳（timestamp）、设备MAC地址（macAddress）、所在区域（zone）的原始数据，统计人员的区域流动路径
• 核心规则：
  • 同一MAC地址下，只有当区域发生变化时，才将新区域加入路径数组
  • 若当前记录与同MAC的上一条记录时间间隔超过1小时，需生成新的统计行
• 最终输出每行包含：start（该段路径的起始时间）、end（该段路径的结束时间）、path（按顺序排列的区域路径数组）

原始示例数据 #

date,   macAddress,         zone
8h10m,  00-B0-D0-63-C2-26, room1
8h12m,  00-B0-D0-63-C2-26, hall
8h15m,  00-A0-B0-23-T2-22, room1
8h16m,  00-A0-B0-23-T2-22, meeting2
8h18m,  00-B0-D0-63-C2-26, meeting2
8h25m,  00-A0-B0-23-T2-22, cafetaria
8h30m,  00-G5-A8-44-T2-30, room1
8h34m,  00-G5-A8-44-T2-30, meeting2
8h49m,  00-G5-A8-44-T2-30, meeting2
14h05m, 00-G5-A8-44-T2-30, cafetaria
14h15m, 00-G5-A8-44-T2-30, room1

期望输出结果 #

macAddress,           start   end     path
00-B0-D0-63-C2-26,    8h10m,  8h18m,  [room1, hall, meeting2]
00-A0-B0-23-T2-22,    8h15m,  8h25m,  [room1, meeting2, cafetaria]
00-G5-A8-44-T2-30,    8h30m,  8h49m,  [room1, meeting2]
00-G5-A8-44-T2-30,    14h05m, 14h15m, [cafetaria, room1]

实现SQL代码 #

WITH processed_data AS (
  -- 转换时间格式并标记新会话触发条件
  SELECT
    macAddress,
    zone,
    timestamp,
    PARSE_TIMESTAMP('%Hh%Mm', timestamp) AS ts,
    -- 计算与同MAC上一条记录的时间间隔（小时）
    TIMESTAMP_DIFF(
      PARSE_TIMESTAMP('%Hh%Mm', timestamp),
      LAG(PARSE_TIMESTAMP('%Hh%Mm', timestamp)) OVER (PARTITION BY macAddress ORDER BY timestamp),
      HOUR
    ) AS time_diff,
    -- 标记是否开启新会话：第一条记录或时间间隔超1小时
    CASE
      WHEN LAG(timestamp) OVER (PARTITION BY macAddress ORDER BY timestamp) IS NULL THEN 1
      WHEN TIMESTAMP_DIFF(
        PARSE_TIMESTAMP('%Hh%Mm', timestamp),
        LAG(PARSE_TIMESTAMP('%Hh%Mm', timestamp)) OVER (PARTITION BY macAddress ORDER BY timestamp),
        HOUR
      ) > 1 THEN 1
      ELSE 0
    END AS new_session_flag
  FROM
    `your-project.your-dataset.your-table` -- 替换为你的实际表名
),
session_groups AS (
  -- 为每个连续会话分配唯一ID
  SELECT
    *,
    SUM(new_session_flag) OVER (PARTITION BY macAddress ORDER BY timestamp) AS session_id
  FROM
    processed_data
),
distinct_zones AS (
  -- 过滤会话内连续重复的区域
  SELECT
    macAddress,
    session_id,
    zone,
    timestamp,
    ts,
    CASE
      WHEN LAG(zone) OVER (PARTITION BY macAddress, session_id ORDER BY timestamp) = zone THEN 0
      ELSE 1
    END AS zone_change_flag
  FROM
    session_groups
),
path_elements AS (
  -- 仅保留区域变化的记录（含会话第一条）
  SELECT
    macAddress,
    session_id,
    zone,
    timestamp,
    ts
  FROM
    distinct_zones
  WHERE
    zone_change_flag = 1
)
-- 聚合生成最终结果
SELECT
  macAddress,
  MIN(timestamp) AS start,
  MAX(timestamp) AS end,
  ARRAY_AGG(zone ORDER BY ts) AS path
FROM
  path_elements
GROUP BY
  macAddress,
  session_id
ORDER BY
  macAddress,
  start;

关键步骤解释 #

1. processed_data：将原始的时间字符串转换为BigQuery可计算的TIMESTAMP类型，同时计算每条记录与同MAC上一条记录的时间差，标记是否需要开启新的会话。
2. session_groups：通过累加new_session_flag，为同一MAC下的连续会话分配唯一的session_id，确保时间间隔超过1小时的记录被分到不同会话组。
3. distinct_zones：标记同一会话内连续重复的区域，避免路径数组中出现重复的连续区域（比如连续两条meeting2只保留一次）。
4. path_elements：筛选出区域发生变化的记录，确保路径数组只包含实际的区域切换节点。
5. 最终聚合：按macAddress和session_id分组，提取会话的起始/结束时间，并用ARRAY_AGG按时间顺序生成路径数组。

注意事项 #

• 如果原始时间戳包含日期信息，需要调整PARSE_TIMESTAMP的格式，例如时间格式为2024-05-20 8h10m时，格式字符串改为'%Y-%m-%d %Hh%Mm'。
• 务必将代码中的your-project.your-dataset.your-table替换为你实际的BigQuery表名。

内容的提问来源于stack exchange，提问作者Diogo Magalhães