如何分析GA4至BigQuery的流式数据并实时更新用户浏览字段

针对你需要从GA4的BigQuery _intraday表实时更新用户画像字段的需求，以下是几个可行的方案：

方案1：定时查询+合并更新（准实时，最低5分钟间隔） #

利用BigQuery的定时查询功能，定期从_intraday表提取最新产品浏览数据，合并到用户画像表中，保留每个用户最近3个浏览产品。

步骤： #

1. 确保你有存储用户画像的目标表（例如user_profiles），包含user_pseudo_id和recent_3_products（数组类型）字段。
2. 创建定时查询，设置执行间隔（比如5分钟），执行以下MERGE逻辑：

MERGE INTO `your-project.your-dataset.user_profiles` AS target
USING (
  SELECT
    user_pseudo_id,
    -- 提取最近3个不重复的浏览产品，按时间倒序
    ARRAY(
      SELECT DISTINCT item_name
      FROM UNNEST(ARRAY_AGG(item_name ORDER BY event_timestamp DESC))
      LIMIT 3
    ) AS latest_recent_products
  FROM `your-project.your-dataset.ga4_intraday_*`
  WHERE 
    event_name = 'view_item'
    -- 只取上次查询后新增的数据，减少计算量
    AND event_timestamp > TIMESTAMP_SUB(CURRENT_TIMESTAMP(), INTERVAL 5 MINUTE)
  GROUP BY user_pseudo_id
) AS source
ON target.user_pseudo_id = source.user_pseudo_id
WHEN MATCHED THEN
  UPDATE SET 
    recent_3_products = ARRAY(
      SELECT DISTINCT *
      FROM UNNEST(ARRAY_CONCAT(source.latest_recent_products, target.recent_3_products))
      ORDER BY event_timestamp DESC LIMIT 3
    )
WHEN NOT MATCHED THEN
  INSERT (user_pseudo_id, recent_3_products)
  VALUES (source.user_pseudo_id, source.latest_recent_products)

3. 配置查询失败告警，确保更新流程稳定。

方案2：物化视图（准实时，最低1分钟间隔） #

用BigQuery的物化视图自动刷新，结合_intraday表和用户画像表，生成实时的用户画像视图，可直接用于后续个性化体验分析。

步骤： #

1. 创建物化视图，设置1分钟刷新间隔：

CREATE MATERIALIZED VIEW `your-project.your-dataset.user_profiles_realtime`
OPTIONS (refresh_interval_minutes = 1)
AS
SELECT
  COALESCE(up.user_pseudo_id, iu.user_pseudo_id) AS user_pseudo_id,
  -- 合并历史画像和最新_intraday数据，保留最近3个产品
  ARRAY(
    SELECT DISTINCT item_name
    FROM UNNEST(
      ARRAY_CONCAT(
        (SELECT ARRAY_AGG(item_name ORDER BY event_timestamp DESC LIMIT 3) 
         FROM `your-project.your-dataset.ga4_intraday_*` iu_sub
         WHERE iu_sub.user_pseudo_id = COALESCE(up.user_pseudo_id, iu.user_pseudo_id)
           AND iu_sub.event_name = 'view_item'),
        up.recent_3_products
      )
    )
    ORDER BY event_timestamp DESC LIMIT 3
  ) AS recent_3_products
FROM `your-project.your-dataset.ga4_intraday_*` iu
FULL OUTER JOIN `your-project.your-dataset.user_profiles` up
ON iu.user_pseudo_id = up.user_pseudo_id
WHERE iu.event_name = 'view_item'
GROUP BY COALESCE(up.user_pseudo_id, iu.user_pseudo_id), up.recent_3_products

2. 直接查询这个物化视图获取实时更新的用户画像数据。

方案3：GA4数据流转Pub/Sub+Cloud Function（真正实时） #

如果需要完全实时（数据到达后立即处理），可以调整GA4的数据流路径，先将数据发送到Pub/Sub，再同步到BigQuery，同时触发Cloud Function处理数据更新用户画像。

步骤： #

1. 重新配置GA4的导出，将数据发送到Cloud Pub/Sub主题。
2. 创建Cloud Function，订阅该Pub/Sub主题，编写代码解析GA4事件数据，提取用户ID和浏览产品信息。
3. 在Cloud Function中调用BigQuery API，更新用户画像表中的recent_3_products字段：
   • 对于已存在的用户：合并现有产品列表与新浏览产品，去重后保留最近3个。
   • 对于新用户：直接插入用户ID和当前浏览产品。

代码示例（Python）： #

import json
from google.cloud import bigquery

bq_client = bigquery.Client()

def update_user_profile(event, context):
    pubsub_message = json.loads(event['data'].decode('utf-8'))
    user_id = pubsub_message['user_pseudo_id']
    item_name = next(p['value']['string_value'] for p in pubsub_message['event_params'] if p['key'] == 'item_name')
    event_timestamp = pubsub_message['event_timestamp']

    # 查询用户现有画像
    query = f"""
    SELECT recent_3_products FROM `your-project.your-dataset.user_profiles`
    WHERE user_pseudo_id = '{user_id}'
    """
    query_job = bq_client.query(query)
    results = query_job.result()

    if results.total_rows > 0:
        # 合并并更新
        existing_products = list(results)[0]['recent_3_products']
        # 插入新产品到头部，去重后保留3个
        new_products = list(dict.fromkeys([item_name] + existing_products))[:3]
        update_query = f"""
        UPDATE `your-project.your-dataset.user_profiles`
        SET recent_3_products = {json.dumps(new_products)}
        WHERE user_pseudo_id = '{user_id}'
        """
    else:
        # 插入新用户
        insert_query = f"""
        INSERT INTO `your-project.your-dataset.user_profiles` (user_pseudo_id, recent_3_products)
        VALUES ('{user_id}', {json.dumps([item_name])})
        """
        update_query = insert_query

    bq_client.query(update_query).result()

注意事项 #

• 去重逻辑：务必处理重复的产品浏览事件，避免列表中出现重复产品。
• 性能优化：对于大规模数据，优先使用分区表、聚类表，查询时限制时间范围减少计算量。
• 成本控制：定时查询和物化视图会产生计算费用，需根据实际需求调整刷新间隔。

内容的提问来源于stack exchange，提问作者Szafirek