如何在ADF中实现验证检查？含数据插入失败自动修复需求示例请求

1. 核心管道架构 #

使用ADF的Try-Catch容器作为核心框架：

• Try块：执行数据插入+验证逻辑，失败时主动抛错
• Catch块：捕获错误后自动修复问题，完成后让管道继续执行后续流程

2. Try块：数据插入与验证检查 #

2.1 带验证的插入操作 #

用Stored Procedure活动执行插入，存储过程内置数据验证逻辑，失败时抛出明确错误（以SQL Server为例）：

CREATE PROCEDURE InsertTargetData
@InputData NVARCHAR(MAX),
@BatchId INT
AS
BEGIN
    BEGIN TRY
        -- 验证必填字段
        IF EXISTS(SELECT 1 FROM OPENJSON(@InputData) WHERE JSON_VALUE(value, '$.Id') IS NULL)
            THROW 50001, '批次【'+CAST(@BatchId AS NVARCHAR)+'】存在缺失Id的记录', 1;
        -- 验证数据格式
        IF EXISTS(SELECT 1 FROM OPENJSON(@InputData) WHERE ISNUMERIC(JSON_VALUE(value, '$.Value')) = 0)
            THROW 50002, '批次【'+CAST(@BatchId AS NVARCHAR)+'】存在数值格式错误的记录', 1;

        -- 执行插入
        INSERT INTO TargetTable (Id, Name, Value, BatchId)
        SELECT Id, Name, Value, @BatchId
        FROM OPENJSON(@InputData)
        WITH (
            Id INT,
            Name NVARCHAR(50),
            Value DECIMAL(18,2)
        );
    END TRY
    BEGIN CATCH
        -- 将错误记录写入临时表，方便后续修复
        INSERT INTO ErrorLogTable (BatchId, ErrorMsg, ErrorTime)
        VALUES (@BatchId, ERROR_MESSAGE(), GETDATE());
        -- 抛出错误给ADF捕获
        THROW;
    END CATCH
END

2.2 插入结果验证 #

在存储过程活动后添加Lookup活动，验证插入数量是否符合预期：

• Lookup查询语句：SELECT COUNT(*) AS InsertCount FROM TargetTable WHERE BatchId = @pipeline().parameters.BatchId
• 后续添加If Condition活动，判断@activity('Lookup_InsertCheck').output.firstRow.InsertCount是否等于@pipeline().parameters.ExpectedCount
• 若不相等，用Fail活动抛出错误：批次【@{pipeline().parameters.BatchId}】插入数量不符，预期@{pipeline().parameters.ExpectedCount}条，实际@{activity('Lookup_InsertCheck').output.firstRow.InsertCount}条

3. Catch块：错误捕获与自动修复 #

3.1 捕获错误信息 #

用Set Variable活动提取错误详情，将变量ErrorDetails赋值为：@activity('InsertData_SP').error.message

3.2 分支处理修复逻辑 #

添加Switch活动，根据错误代码分支处理：

• 分支1（错误代码50001：缺失Id）：调用修复存储过程补充Id后重新插入

  CREATE PROCEDURE FixMissingIdRecords
  @BatchId INT
  AS
  BEGIN
      -- 为错误记录生成自增Id（根据业务调整逻辑）
      UPDATE ErrorTempTable
      SET Id = (SELECT ISNULL(MAX(Id),0)+ROW_NUMBER() OVER(ORDER BY Name) FROM TargetTable)
      WHERE BatchId = @BatchId;
  
      -- 重新插入修复后的数据
      INSERT INTO TargetTable (Id, Name, Value, BatchId)
      SELECT Id, Name, Value, BatchId
      FROM ErrorTempTable
      WHERE BatchId = @BatchId;
  END
  

• 分支2（错误代码50002：数值格式错误）：调用存储过程清洗数据格式

  CREATE PROCEDURE FixInvalidValueRecords
  @BatchId INT
  AS
  BEGIN
      -- 将非数值内容替换为默认值0（根据业务调整）
      UPDATE ErrorTempTable
      SET Value = 0
      WHERE BatchId = @BatchId AND ISNUMERIC(Value) = 0;
  
      -- 重新插入修复后的数据
      INSERT INTO TargetTable (Id, Name, Value, BatchId)
      SELECT Id, Name, Value, BatchId
      FROM ErrorTempTable
      WHERE BatchId = @BatchId;
  END
  

• 默认分支：记录未处理错误到日志表，避免遗漏

3.3 标记修复状态 #

修复完成后，用Set Variable活动将FixStatus设为Success，确保管道能正常退出Catch块

4. 管道续跑配置 #

将需要后续执行的活动（如数据汇总、通知）直接放在Try-Catch容器之后，Catch块处理完修复后，管道会自动继续执行后续流程

5. 关键注意事项 #

• 确保数据库账号拥有存储过程的执行、表的增删改权限
• 错误日志表需包含批次ID、错误信息、修复状态等字段，便于后续排查
• 可在Catch块末尾添加Web活动发送修复通知，同步状态给业务方

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