Azure Data Factory v2如何基于时间戳增量复制Data Lake目录至SQL Server

我来帮你梳理下这个场景的完整解决方案，既能满足历史数据加载需求，又能通过Lookup活动实现后续的增量同步，完全适配你用时间戳命名目录的结构：

整体流程逻辑 #

核心思路很清晰：先用Lookup活动从SQL Server读取上次复制完成的最大时间戳，接着在复制活动的ADLS源数据集里用过滤器筛选出大于该时间戳的目录/文件，完成复制后再更新SQL里的时间戳记录，形成闭环的增量同步流程。

步骤1：准备SQL时间戳记录表 #

首先得在你的SQL Server里建一张专门记录复制历史的表，用来存每次复制的最大时间戳。初始插入一个极小值（比如0），这样第一次运行就能加载所有历史数据：

CREATE TABLE CopyHistory (
    Id INT IDENTITY(1,1) PRIMARY KEY,
    LastCopiedTimestamp BIGINT NOT NULL,
    CopyCompletedAt DATETIME DEFAULT GETDATE()
);
-- 初始值设为0，确保第一次运行能拉取所有历史目录
INSERT INTO CopyHistory (LastCopiedTimestamp) VALUES (0);

步骤2：用Lookup获取上次复制的时间戳 #

在ADF管道里加一个Lookup活动，配置它连接到你的SQL Server数据集，查询语句用：

SELECT MAX(LastCopiedTimestamp) AS LastTimestamp FROM CopyHistory;

这个活动的输出会包含上次复制的最大时间戳，后续的复制活动可以直接引用这个值。

步骤3：配置ADLS源的过滤规则（核心步骤） #

针对你的时间戳目录场景，有两种过滤方式可选，按需选择：

方式一：按目录名过滤（推荐，匹配你的目录结构） #

在ADLS源数据集的过滤器配置里，用ADF的逻辑函数判断目录名是否大于Lookup返回的时间戳。假设你的目录是ingest/{timestamp}/格式，过滤器表达式写：
@greater(int(item().name), int(activity('LookupLastTimestamp').output.firstRow.LastTimestamp))

• item().name 指当前遍历到的子目录名称（也就是那个时间戳字符串）
• 先把它转成int类型，再和Lookup拿到的上次时间戳（同样转int）做比较，只保留大于该值的目录

方式二：按文件名过滤（如果文件名带时间戳） #

如果你的文件名也包含时间戳（比如file_1510395023.tsv），可以调整过滤器表达式：
@greater(int(split(item().name, '_')[1]), int(activity('LookupLastTimestamp').output.firstRow.LastTimestamp))
这里用split拆分文件名，提取出时间戳部分再转成int做比较。

划重点：第一次运行时，因为SQL表里的初始值是0，所有时间戳目录都会被选中，完美满足你加载历史数据的需求。

步骤4：配置复制活动 #

把复制活动的源设为刚才配置好过滤器的ADLS数据集，目标设为你的SQL Server数据集。复制模式推荐用追加（如果是新增数据），或者根据业务需求选择更新模式。记得开启ADLS数据集的递归选项，这样才能遍历所有子目录下的文件。

步骤5：更新SQL的时间戳记录 #

复制完成后，需要把本次复制的最大时间戳更新到SQL表里，确保下次运行只拉取新的目录。这里分两步：

1. 再加一个Lookup活动，获取ADLS里ingest目录下的最大时间戳：

SELECT MAX(CAST(name AS BIGINT)) AS CurrentMaxTimestamp 
FROM OPENROWSET(
    BULK 'ingest/*',
    DATA_SOURCE = 'YourADLSDataSource', -- 替换成你的ADLS数据源名称
    SINGLE_CLOB
) AS Directories
WHERE ISNUMERIC(name) = 1;

2. 用Stored Procedure活动调用SQL存储过程，把这个最大时间戳写入CopyHistory表：

CREATE PROCEDURE UpdateCopyHistory @NewTimestamp BIGINT
AS
BEGIN
    -- 插入新记录，保留所有复制历史
    INSERT INTO CopyHistory (LastCopiedTimestamp) VALUES (@NewTimestamp);
    -- 如果只想保留最新一条，也可以用UPDATE：
    -- UPDATE CopyHistory SET LastCopiedTimestamp = @NewTimestamp WHERE Id = (SELECT MAX(Id) FROM CopyHistory);
END

在Stored Procedure活动里，把Lookup获取的CurrentMaxTimestamp作为参数传递进去就行。

额外优化建议 #

• 如果时间戳是13位毫秒级（比如1690000000000），一定要用bigint类型，避免int溢出
• 可以加一个If Condition活动，判断本次是否有新目录：如果Lookup返回的CurrentMaxTimestamp等于上次的LastTimestamp，就跳过复制步骤，节省资源
• 测试时可以先手动修改SQL表里的LastCopiedTimestamp，验证过滤逻辑是否生效

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