如何用Azure Data Factory获取SQL表中未同步到ADLS的新增数据

为什么Join转换没得到预期结果？ #

你之前用的应该是INNER JOIN，这种Join只会返回两张表中匹配的行，而你需要的是SQL源表中独有的新增行，所以得换用左连接筛选或存在性判断的逻辑。

方法1：存在性筛选（通用无依赖） #

如果ADLS中的备份数据可以通过查询访问（比如导入临时SQL表、创建ADLS外部表），可以用以下两种方式筛选新增行：

示例1：使用NOT EXISTS #

假设：

• SQL源表：source_sql_table
• ADLS备份对应的查询/临时表：adls_backup_table
• 用唯一键（如id）匹配行

SELECT s.*
FROM source_sql_table s
WHERE NOT EXISTS (
    SELECT 1
    FROM adls_backup_table b
    WHERE s.id = b.id
);

示例2：使用LEFT JOIN+NULL筛选 #

SELECT s.*
FROM source_sql_table s
LEFT JOIN adls_backup_table b ON s.id = b.id
WHERE b.id IS NULL;

方法2：利用唯一标识/时间戳（高效增量） #

如果你的SQL表有自增主键或记录插入时间的列，可以直接基于第一次复制的同步点筛选，效率远高于全表对比：

示例：基于自增ID #

假设第一次复制时，SQL表的最大id为1000，后续新增行的id均大于该值：

SELECT *
FROM source_sql_table
WHERE id > 1000;

示例：基于插入时间 #

假设第一次复制完成时间为2024-05-20 12:00:00，筛选该时间后插入的行：

SELECT *
FROM source_sql_table
WHERE created_at > '2024-05-20 12:00:00';

方法3：记录同步点（长期增量同步方案） #

如果需要持续同步新增数据，建议维护一个元数据表记录同步断点，每次同步时读取断点、筛选数据、更新断点：

1. 创建元数据表：

CREATE TABLE sync_metadata (
    table_name VARCHAR(100) PRIMARY KEY,
    last_sync_id INT,
    last_sync_time DATETIME
);

2. 第一次复制后写入初始同步点：

INSERT INTO sync_metadata (table_name, last_sync_id, last_sync_time)
VALUES ('source_sql_table', 1000, '2024-05-20 12:00:00');

3. 获取新增行：

SELECT s.*
FROM source_sql_table s
JOIN sync_metadata m ON m.table_name = 'source_sql_table'
WHERE s.id > m.last_sync_id;

4. 同步完成后更新同步点：

UPDATE sync_metadata
SET last_sync_id = (SELECT MAX(id) FROM source_sql_table)
WHERE table_name = 'source_sql_table';

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