如何从DataLake取JSON在Azure Synapse专用SQL池实现条件Upsert？

核心思路 #

先把Data Lake中的JSON数据加载到Synapse临时表，再通过Synapse的MERGE语句实现增量同步：以id字段匹配判断记录是否存在，对比lastmodified时间戳决定执行更新、忽略或插入操作。

具体实现步骤 #

1. 准备目标表（已存在可跳过） #

创建Synapse中存储数据的目标表，包含作为主键的id、用于判断更新的lastmodified及业务字段：

CREATE TABLE dbo.target_table (
    id VARCHAR(50) PRIMARY KEY,
    lastmodified DATETIME2 NOT NULL,
    -- 示例业务字段，根据实际JSON结构调整
    name VARCHAR(100),
    value INT
);

2. 将Data Lake的JSON加载到临时表 #

用OPENROWSET读取ADLS Gen2中的JSON文件，导入临时表（也可选用COPY INTO，依据权限场景选择）：

-- 创建临时表存储源JSON数据
CREATE TABLE #staging_json (
    id VARCHAR(50),
    lastmodified DATETIME2,
    name VARCHAR(100),
    value INT
);

-- 从ADLS读取JSON并插入临时表
INSERT INTO #staging_json
SELECT
    JSON_VALUE(json_data, '$.id') AS id,
    JSON_VALUE(json_data, '$.lastmodified') AS lastmodified,
    JSON_VALUE(json_data, '$.name') AS name,
    JSON_VALUE(json_data, '$.value') AS value
FROM OPENROWSET(
    BULK 'https://youradlsaccount.dfs.core.windows.net/yourcontainer/path/*.json',
    FORMAT = 'CSV',
    FIELDTERMINATOR = '0x0b',
    FIELDQUOTE = '0x0b',
    ROWTERMINATOR = '0x0a'
) WITH (json_data VARCHAR(MAX)) AS rows;

注意：若JSON为数组格式，可结合OPENJSON解析；根据实际嵌套结构调整JSON_VALUE的路径。

3. 执行增量Upsert（MERGE） #

通过MERGE语句实现核心逻辑：

MERGE INTO dbo.target_table AS target
USING #staging_json AS source
ON target.id = source.id

-- id匹配且源数据更新时间更晚时，更新记录
WHEN MATCHED AND source.lastmodified > target.lastmodified THEN
    UPDATE SET
        target.lastmodified = source.lastmodified,
        target.name = source.name,
        target.value = source.value

-- id不匹配时，插入新记录
WHEN NOT MATCHED THEN
    INSERT (id, lastmodified, name, value)
    VALUES (source.id, source.lastmodified, source.name, source.value);

说明：若源数据lastmodified早于目标表记录，MERGE会自动忽略该条，无需额外处理。

示例流程验证 #

假设目标表初始数据：

Data Lake的JSON文件内容：

{"id":"001","lastmodified":"2024-01-02 12:00:00","name":"Alice","value":200}
{"id":"002","lastmodified":"2024-01-02 13:00:00","name":"Bob","value":150}
{"id":"001","lastmodified":"2023-12-30 09:00:00","name":"Alice","value":50}

执行上述步骤后，目标表最终数据：

解释：id=001的旧时间戳记录被忽略，新时间戳记录完成更新；id=002的新记录成功插入。

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