如何通过ADF将Google Analytics v4 JSON动态转换为SQL表

GA v4 Reporting API返回的是列元数据与数据行分离的报告结构，没有在每条数据行上重复标注字段名，核心处理逻辑是先提取统一声明的列头信息，再按顺序对齐每行的维度、指标值，以下三种方案均可在ADF链路中落地，可根据自身场景选择：

方案1：ADF映射数据流原生实现（推荐，无额外服务依赖） #

全程用ADF内置组件完成，不需要额外部署代码服务，步骤如下：

• 源端配置：源选择REST连接器/JSON文件数据集，直接对接GA接口返回结果。首先在管道层面提取列头元数据存为变量：
  • 维度列名数组：直接取返回体中columnHeader.dimensions的数组值
  • 指标列名数组：遍历columnHeader.metricHeader.metricHeaderEntries，提取每个元素的name字段拼成数组
    合并两个数组得到完整列名列表，同时记录每个指标对应的type字段，用于后续类型转换。
• 行数据打平：在映射数据流中对data.rows数组做两次展开：第一次展开metrics数组（常规非透视请求下该数组每个行只有1个元素），拿到单条记录的指标values数组；第二次将每行的dimensions数组和指标values数组合并，得到和列名列表顺序完全一致的值数组。
• 动态列生成：用派生列转换，循环遍历之前存储的列名数组，按索引位置从值数组中取对应值，动态生成同名字段，全程不需要硬编码列名，后续GA请求调整了维度、指标字段也不需要修改流程。
• 写入SQL：接收器选择目标SQL库，根据之前记录的指标类型做字段类型转换（INTEGER转INT/BIGINT，TIME转数值存秒数即可），开启自动表映射即可写入。如果是首次同步可以开自动建表，后续增量同步直接映射对应字段。

注意：返回体中的totals、samplesReadCounts、samplingSpaceSizes属于聚合/抽样元数据，不要和明细行一起展开，建议单独存一张同步日志表，避免明细数据重复。

方案2：ADF+轻量脚本预处理（灵活度最高，适配所有GA返回结构） #

如果数据流对动态数组的处理有兼容问题，可以加一层极轻量的预处理逻辑，适配性最强：

• ADF拿到GA接口返回的JSON后，直接传给内置的Python/PowerShell活动，或者绑定Azure Function做处理。
• 脚本逻辑非常简单，核心就是做键值对映射：
  1. 读取JSON内容，分别提取维度列名、指标列名，合并为完整表头列表
  2. 遍历data.rows数组，把每行的dimensions值、metrics[0].values值按顺序和表头列表一一对应，转成「字段名:字段值」格式的常规JSON对象数组，也就是普通结构、每条数据自带列名的JSON格式
  3. 把转换后的标准JSON传回ADF复制活动，后续就可以按普通JSON的导入流程直接映射写入SQL，不需要做特殊配置。
• 核心转换代码参考（Python）：

def transform_ga_raw(raw_content):
    report = raw_content[0]
    # 提取列头
    dim_cols = report["columnHeader"]["dimensions"]
    metric_cols = [i["name"] for i in report["columnHeader"]["metricHeader"]["metricHeaderEntries"]]
    all_cols = dim_cols + metric_cols
    # 转换明细行
    detail_rows = []
    for row in report["data"]["rows"]:
        row_values = row["dimensions"] + row["metrics"][0]["values"]
        detail_rows.append(dict(zip(all_cols, row_values)))
    # 提取同步元数据单独存储
    sync_meta = {
        "row_count": report["data"]["rowCount"],
        "is_sampled": True if report["data"].get("samplesReadCounts") else False,
        "sample_read": report["data"].get("samplesReadCounts", [None])[0],
        "sample_total": report["data"].get("samplingSpaceSizes", [None])[0]
    }
    return detail_rows, sync_meta

不管GA请求里加了多少维度、指标，这个脚本都能自动适配，不需要修改逻辑。

方案3：SQL侧直接解析JSON（适合小数据量场景，无额外计算资源消耗） #

如果同步数据量不大，也可以直接把原始JSON传到SQL侧，用数据库内置JSON函数解析：

• 先把GA返回的完整JSON存入SQL的临时变量，或者用OPENROWSET直接读取ADF落地到存储的JSON文件。
• 动态拼接解析SQL：先从columnHeader路径提取所有列名，再遍历data.rows数组，按数组索引对应列名取值，直接插入目标表，全程不需要硬编码字段。
• Azure SQL/SQL Server核心解析逻辑参考：

DECLARE @ga_raw NVARCHAR(MAX) = '/* 此处替换为GA返回的完整JSON内容 */';
DECLARE @col_select NVARCHAR(MAX), @exec_sql NVARCHAR(MAX);

-- 动态生成列选择逻辑
WITH cols AS (
    SELECT [key] AS idx, [value] AS col_name, 0 AS col_type 
    FROM OPENJSON(@ga_raw, '$[0].columnHeader.dimensions')
    UNION ALL
    SELECT 
        (SELECT COUNT(*) FROM OPENJSON(@ga_raw, '$[0].columnHeader.dimensions')) + [key] AS idx,
        JSON_VALUE(val, '$.name') AS col_name, 
        1 AS col_type
    FROM OPENJSON(@ga_raw, '$[0].columnHeader.metricHeader.metricHeaderEntries') val
)
SELECT @col_select = STRING_AGG(
    CASE WHEN col_type = 0 THEN
        CONCAT('JSON_VALUE(row_arr.value, ''$.dimensions[', idx, ']'') AS [', col_name, ']')
    ELSE
        CONCAT('JSON_VALUE(metric_val.value, ''$.values[', idx - (SELECT COUNT(*) FROM cols WHERE col_type=0), ']'') AS [', col_name, ']')
    END,
    ','
)
FROM cols;

-- 执行解析写入
SET @exec_sql = CONCAT('
INSERT INTO dbo.ga_source_report
SELECT ', @col_select, '
FROM OPENJSON(@ga_raw, ''$[0].data.rows'') row_arr
CROSS APPLY OPENJSON(row_arr.value, ''$.metrics'') metric_arr
CROSS APPLY OPENJSON(metric_arr.value) metric_val
');
EXEC sp_executesql @exec_sql, N'@ga_raw NVARCHAR(MAX)', @ga_raw = @ga_raw;

这个方案不需要额外配置其他服务，但单批次数据超过10万行时解析性能较差，适合小批量定时同步场景。

落地注意事项 #

• GA返回的所有指标值默认是字符串格式，写入SQL前必须根据metricHeaderEntries里的type字段做显式类型转换，避免出现类型不兼容报错。
• 如果单次请求返回的rowCount超过10万，要在ADF里配置GA接口分页拉取，不要一次性拉取全量数据导致内存溢出。
• 维度字段里可能出现特殊字符，写入SQL时注意字段名的转义，避免语法错误。

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