通过SQL存储过程调用REST API实现ADF管道状态轮询

逻辑思路 #

要替代ADF的Wait-Until组件实现低成本轮询，存储过程需要完成以下核心步骤：

• 接收ADF工厂名、目标管道运行ID、重试间隔、超时时间等参数
• 循环调用ADF管道运行状态查询接口
• 根据返回状态分支处理：
  • 若状态为'InProgress'，等待指定间隔后重新查询
  • 若状态变为'Succeeded'或'Failed'，结束循环并返回最终状态
  • 超过预设超时时间则抛出超时异常

存储过程实现（Azure SQL Database） #

以下是基于Azure SQL DB的存储过程示例，利用sp_invoke_external_rest_endpoint调用ADF状态查询接口。需提前确保SQL DB已启用外部REST调用权限，且托管标识拥有ADF管道运行的查询权限：

CREATE PROCEDURE dbo.ADF_PollPipelineRunStatus
    @FactoryName NVARCHAR(255),
    @PipelineRunId NVARCHAR(100),
    @RetryIntervalSeconds INT = 30, -- 默认30秒重试一次
    @TimeoutSeconds INT = 3600 -- 默认超时1小时
AS
BEGIN
    SET NOCOUNT ON;

    DECLARE @StartTime DATETIME = GETDATE();
    DECLARE @TimeoutDeadline DATETIME = DATEADD(SECOND, @TimeoutSeconds, @StartTime);
    DECLARE @ApiEndpoint NVARCHAR(500);
    DECLARE @ApiResponse NVARCHAR(MAX);
    DECLARE @CurrentRunStatus NVARCHAR(50);
    -- 获取托管标识的访问令牌用于ADF API授权
    DECLARE @AuthHeader NVARCHAR(MAX) = N'{"Authorization": "Bearer ' + (SELECT CONVERT(NVARCHAR(MAX), access_token) FROM sys.dm_azuredb_access_token WHERE resource = 'https://management.azure.com/') + '"}';

    -- 构造ADF状态查询API地址，替换为你的订阅ID和资源组名
    SET @ApiEndpoint = N'https://management.azure.com/subscriptions/[你的订阅ID]/resourceGroups/[你的资源组名]/providers/Microsoft.DataFactory/factories/' + @FactoryName + '/pipelineruns/' + @PipelineRunId + '?api-version=2018-06-01';

    WHILE GETDATE() < @TimeoutDeadline
    BEGIN
        -- 调用ADF API获取当前运行状态
        EXEC sp_invoke_external_rest_endpoint
            @url = @ApiEndpoint,
            @method = 'GET',
            @headers = @AuthHeader,
            @response = @ApiResponse OUTPUT;

        -- 从JSON响应中解析状态字段
        SET @CurrentRunStatus = JSON_VALUE(@ApiResponse, '$.properties.status');

        -- 状态判断逻辑
        IF @CurrentRunStatus IN ('Succeeded', 'Failed')
        BEGIN
            SELECT @CurrentRunStatus AS FinalRunStatus;
            RETURN;
        END
        ELSE IF @CurrentRunStatus = 'InProgress'
        BEGIN
            -- 等待指定间隔后重试
            WAITFOR DELAY = CONVERT(VARCHAR(8), DATEADD(SECOND, @RetryIntervalSeconds, 0), 108);
        END
        ELSE
        BEGIN
            -- 未知状态抛出错误
            THROW 50001, N'检测到未知的管道运行状态', 1;
        END
    END

    -- 超时抛出错误
    THROW 50002, N'管道状态轮询超时', 1;
END

ADF集成步骤 #

1. 在ADF管道中添加存储过程活动，关联托管该存储过程的Azure SQL数据库
2. 指定存储过程名称为dbo.ADF_PollPipelineRunStatus
3. 配置参数：
   • @FactoryName：使用ADF系统变量@pipeline().DataFactory获取当前工厂名
   • @PipelineRunId：从前置的Execute Pipeline活动输出中获取，如@activity('执行目标管道').output.runId
   • @RetryIntervalSeconds：根据业务需求设置，如30秒
   • @TimeoutSeconds：设置最大等待时长，如3600秒（1小时）
4. 后续可通过If Condition活动，根据存储过程返回的FinalRunStatus分支处理成功/失败逻辑

关键注意事项 #

• 权限配置：确保Azure SQL的托管标识拥有ADF工厂的Monitoring Reader或Data Factory Contributor权限，否则无法调用状态查询接口
• API版本：示例使用的API版本为2018-06-01，可根据实际需求替换为最新版本
• 错误处理：存储过程包含未知状态和超时的异常抛出，需在ADF中为存储过程活动配置对应的错误处理策略（如重试、终止）
• 成本优化：相比Wait-Until组件，该方案利用SQL计算资源实现等待逻辑，长时间轮询场景下成本更低，可根据负载选择合适的SQL DB定价层

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