求基于Azure Java SDK连接Azure Data Factory的Java示例代码

前置依赖（Maven） #

先在pom.xml中添加Azure Data Factory SDK和认证相关依赖：

<dependencies>
    <dependency>
        <groupId>com.azure.resourcemanager</groupId>
        <artifactId>azure-resourcemanager-datafactory</artifactId>
        <version>1.0.0</version> <!-- 建议使用最新稳定版 -->
    </dependency>
    <dependency>
        <groupId>com.azure</groupId>
        <artifactId>azure-identity</artifactId>
        <version>1.12.0</version>
    </dependency>
</dependencies>

核心功能代码实现 #

初始化Data Factory客户端 #

使用服务主体（Client Secret）认证，创建DataFactory管理客户端：

import com.azure.identity.ClientSecretCredential;
import com.azure.identity.ClientSecretCredentialBuilder;
import com.azure.resourcemanager.datafactory.DataFactoryManager;
import com.azure.resourcemanager.datafactory.models.PipelineRun;
import java.util.Collections;

public class ADFPipelineOperations {
    // 替换为你的Azure认证信息和资源信息
    private static final String TENANT_ID = "你的租户ID";
    private static final String CLIENT_ID = "你的客户端ID";
    private static final String CLIENT_SECRET = "你的客户端密钥";
    private static final String SUBSCRIPTION_ID = "你的订阅ID";
    private static final String RESOURCE_GROUP_NAME = "你的资源组名";
    private static final String DATA_FACTORY_NAME = "你的数据工厂名";

    private final DataFactoryManager dataFactoryManager;

    public ADFPipelineOperations() {
        // 创建认证凭据
        ClientSecretCredential credential = new ClientSecretCredentialBuilder()
                .tenantId(TENANT_ID)
                .clientId(CLIENT_ID)
                .clientSecret(CLIENT_SECRET)
                .build();

        // 初始化DataFactoryManager
        this.dataFactoryManager = DataFactoryManager.authenticate(credential, SUBSCRIPTION_ID);
    }

触发管道运行 #

// 触发管道运行，可传入管道参数（可选）
    public String triggerPipeline(String pipelineName) {
        PipelineRun run = dataFactoryManager.pipelines()
                .createRunWithHttpResponse(RESOURCE_GROUP_NAME, DATA_FACTORY_NAME, pipelineName, null)
                .getValue();
        return run.runId();
    }

获取管道运行状态 #

// 根据运行ID获取管道状态
    public String getPipelineRunStatus(String runId) {
        PipelineRun run = dataFactoryManager.pipelineRuns()
                .get(RESOURCE_GROUP_NAME, DATA_FACTORY_NAME, runId);
        return run.status();
    }

取消管道运行 #

// 取消指定运行ID的管道运行
    public void cancelPipelineRun(String runId) {
        dataFactoryManager.pipelineRuns()
                .cancel(RESOURCE_GROUP_NAME, DATA_FACTORY_NAME, runId, "用户主动取消");
    }

完整使用示例 #

在main方法中演示完整流程：

public static void main(String[] args) {
        String pipelineName = "你的管道名称";
        ADFPipelineOperations adfOps = new ADFPipelineOperations();

        // 1. 触发管道运行
        String runId = adfOps.triggerPipeline(pipelineName);
        System.out.println("管道已触发，运行ID：" + runId);

        // 2. 轮询获取运行状态
        try {
            String status;
            do {
                Thread.sleep(5000); // 每5秒查询一次
                status = adfOps.getPipelineRunStatus(runId);
                System.out.println("当前管道状态：" + status);
            } while (!status.equals("Succeeded") && !status.equals("Failed") && !status.equals("Cancelled"));
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
        }

        // 3. 如需取消管道（示例：状态为Running时取消）
        String currentStatus = adfOps.getPipelineRunStatus(runId);
        if ("Running".equals(currentStatus)) {
            adfOps.cancelPipelineRun(runId);
            System.out.println("管道已取消");
        }
    }
}

注意事项 #

• 确保服务主体拥有Data Factory Contributor或对应权限的角色
• 替换代码中所有你的XXX为实际Azure资源信息
• 轮询间隔可根据实际管道运行时长调整
• 管道参数可通过createRunWithHttpResponse方法的第三个参数传入（类型为Map<String, Object>）

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