如何每日将GA4事件数据经BigQuery同步至ADF Synapse Delta表并扁平化嵌套结构？

问题背景 #

我们从GA4提取事件表数据时选择用BigQuery（规避Google API的行数、维度/指标数量限制），但ADF读取后返回多层嵌套JSON格式：

{"v": [{"v": {"f": [{"v": "firebase_conversion"}, {"v": {"f": [{"v": null}, {"v": "0"}, {"v": null}, {"v": null}]}}]}}, {"v": {"f": [{"v": "ga_session_id"}, {"v": {"f": [{"v": null}, {"v": "123"}, {"v": null}, {"v": null}]}}]}}, {"v": {"f": [{"v": "engaged_session_event"}, {"v": {"f": [{"v": null}, {"v": "1"}, {"v": null}, {"v": null}]}}]}}, {"v": {"f": [{"v": "ga_session_number"}, {"v": {"f": [{"v": null}, {"v": "9"}, {"v": null}, {"v": null}]}}]}}, {"v": {"f": [{"v": "page_referrer"}, {"v": {"f": [{"v": "ABC"}, {"v": null}, {"v": null}, {"v": null}]}}]}}, {"v": {"f": [{"v": "page_title"}, {"v": {"f": [{"v": "ABC"}, {"v": null}, {"v": null}, {"v": null}]}}]}}, {"v": {"f": [{"v": "page_location"}, {"v": {"f": [{"v": "xyz"}, {"v": null}, {"v": null}, {"v": null}]}}]}}, {"v": {"f": [{"v": "session_engaged"}, {"v": {"f": [{"v": null}, {"v": "1"}, {"v": null}, {"v": null}]}}]}}]}

用unnest处理嵌套列会导致行数暴增（350万条变为4000万条），尝试先提取原始数据再用Azure Functions的Python函数扁平化，但空值处理困难。需要一套每日同步、无数据膨胀、能在数据湖生成目标格式的最优方案。

最优方案推荐 #

方案1：BigQuery端提前扁平化（首推） #

直接在BigQuery中解析嵌套结构，从根源避免数据膨胀：

• 编写SQL提取每个字段的有效值，自动过滤空值：

  SELECT
    -- 提取firebase_conversion的有效值
    (SELECT item.v.f[1].f[offset(1)].v 
     FROM UNNEST(json_extract_array(raw_data, '$.v')) AS item
     WHERE item.v.f[0].v = 'firebase_conversion') AS firebase_conversion,
    -- 提取ga_session_id的有效值
    (SELECT item.v.f[1].f[offset(1)].v 
     FROM UNNEST(json_extract_array(raw_data, '$.v')) AS item
     WHERE item.v.f[0].v = 'ga_session_id') AS ga_session_id,
    -- 依次处理其他字段
    (SELECT item.v.f[1].f[offset(1)].v 
     FROM UNNEST(json_extract_array(raw_data, '$.v')) AS item
     WHERE item.v.f[0].v = 'engaged_session_event') AS engaged_session_event,
    (SELECT item.v.f[1].f[offset(1)].v 
     FROM UNNEST(json_extract_array(raw_data, '$.v')) AS item
     WHERE item.v.f[0].v = 'ga_session_number') AS ga_session_number,
    (SELECT item.v.f[1].f[offset(0)].v 
     FROM UNNEST(json_extract_array(raw_data, '$.v')) AS item
     WHERE item.v.f[0].v = 'page_referrer') AS page_referrer,
    (SELECT item.v.f[1].f[offset(0)].v 
     FROM UNNEST(json_extract_array(raw_data, '$.v')) AS item
     WHERE item.v.f[0].v = 'page_title') AS page_title,
    (SELECT item.v.f[1].f[offset(0)].v 
     FROM UNNEST(json_extract_array(raw_data, '$.v')) AS item
     WHERE item.v.f[0].v = 'page_location') AS page_location,
    (SELECT item.v.f[1].f[offset(1)].v 
     FROM UNNEST(json_extract_array(raw_data, '$.v')) AS item
     WHERE item.v.f[0].v = 'session_engaged') AS session_engaged
  FROM your_ga4_dataset.target_table
  

  注：raw_data需替换为存储嵌套JSON的实际字段名；不同字段的有效值位置不同（指标类在offset(1)，维度类在offset(0)），按需调整。
• 将上述SQL封装为BigQuery视图，ADF直接读取视图数据，无需处理嵌套结构，行数与原始数据一致。
• 配置ADF每日同步该视图数据到数据湖，完成自动化流程。

方案2：优化Azure Functions的Python扁平化逻辑 #

若必须在Azure端处理，优化空值提取逻辑：

• 编写Python函数遍历嵌套数组，精准提取有效值并跳过空值：

  def flatten_ga4_json(raw_json):
      flat_result = {}
      for entry in raw_json.get('v', []):
          field_name = entry.get('v', {}).get('f', [{}])[0].get('v')
          if not field_name:
              continue
          # 提取值列表，过滤空值
          value_items = entry.get('v', {}).get('f', [{}])[1].get('f', [])
          valid_value = next((item.get('v') for item in value_items if item.get('v') is not None), None)
          flat_result[field_name] = valid_value
      return flat_result
  

• 在ADF中配置批量触发该函数，处理后直接写入数据湖，注意设置合理的批量大小保证性能。

方案3：ADF内置派生列处理 #

利用ADF的派生列功能，通过jsonPath直接提取字段：

• 添加派生列，针对每个字段编写jsonPath表达式，例如：
  • 提取firebase_conversion：jsonPath($, '$.v[?(@.v.f[0].v == "firebase_conversion")].v.f[1].f[1].v')
  • 提取page_referrer：jsonPath($, '$.v[?(@.v.f[0].v == "page_referrer")].v.f[1].f[0].v')
• 所有字段处理完成后，直接将扁平数据写入数据湖，无需额外代码。

方案对比 #

内容的提问来源于stack exchange，提问作者Alefiya S. Tikiwala