在ADF增量复制任务中，如何从Copy Data活动输出获取水印列最大值？

首先直接给结论：默认情况下，Copy Data活动的输出不会返回目标数据集中水印列的最大值。它的输出仅包含复制任务的基础统计信息，比如复制行数、字节数、耗时等，典型的输出结构如下：

{
  "dataRead": 2048,
  "dataWritten": 2048,
  "rowsRead": 150,
  "rowsCopied": 150,
  "copyDuration": 8,
  "sourcePeakConnections": 1,
  "sinkPeakConnections": 1
}

这些信息里没有包含目标数据集的列级聚合值，所以无法直接通过@activity('copyActivity1').output拿到水印列的最大值。

既然不能额外用Lookup查询源表，这里提供两种可行的方案：

方案1：在Copy Data的Source查询中同时计算最大值并传递 #

因为你做的是增量复制，这批数据的水印最大值其实就是源表中大于当前水印的所有记录的最大值。你可以把这个最大值的计算和增量数据查询合并到同一个SQL语句中，然后通过动态内容将其存入管道变量，不需要单独的Lookup活动：

1. 先在管道中定义一个日期类型的变量，比如NewWatermarkValue。
2. 在Copy Data活动的Source选项卡中，选择"Query"模式，使用如下SQL语句（假设你的源表是YourSourceTable，水印列是watermarkDateColumn，当前水印存在管道参数CurrentWatermark中）：

   WITH IncrementalBatch AS (
       SELECT *
       FROM YourSourceTable
       WHERE watermarkDateColumn > @pipeline().parameters.CurrentWatermark
   ),
   MaxWatermarkCalc AS (
       SELECT MAX(watermarkDateColumn) AS MaxWatermark
       FROM IncrementalBatch
   )
   SELECT i.*, m.MaxWatermark
   FROM IncrementalBatch i, MaxWatermarkCalc m
   

   这个查询会给每一行增量数据附加一个MaxWatermark列，所有行的这个值都是这批数据的水印最大值。
3. 在Copy Data活动完成后，使用Set Variable活动，将NewWatermarkValue设置为：

   @if(equals(activity('copyActivity1').output.rowsCopied, 0), variables('NewWatermarkValue'), first(activity('copyActivity1').output.data).MaxWatermark)
   

   这里利用first()函数取第一行的MaxWatermark值（因为所有行的这个值都一样），如果没有复制任何数据（rowsCopied=0），则保持原有值。

方案2：利用ADLS Gen2的分区路径获取最大值 #

如果你的Sink是按watermarkDateColumn分区存储的（比如按日期创建yyyy/mm/dd格式的文件夹），那么最新的分区文件夹名称就是水印列的最大值：

1. 使用Get Metadata活动，指向ADLS Gen2中存储增量数据的根容器/文件夹，勾选"Child items"选项。
2. 在Get Metadata活动完成后，用动态表达式提取最新的分区文件夹名称，比如：

   @max(activity('GetMetadata1').output.childItems[].name)
   

   这个表达式会获取所有子文件夹的名称，然后取最大值（如果分区格式是可排序的日期字符串，比如2024/05/20，这个方法完全可行）。
3. 将这个最大值转换为日期格式，存入你的控制表或管道变量即可。

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