使用Azure Data Factory Java SDK配置ADX源数据集查询过滤

你需要明确：AzureDataExplorerTableDataset仅用于指定源表的基础连接信息，过滤查询是配置在复制活动的源（AzureDataExplorerSource）对象中，而非数据集本身。以下是具体实现步骤和代码示例：

核心实现逻辑 #

• 创建源Kusto数据集，指定表名与关联的链接服务
• 实例化AzureDataExplorerSource，通过专属方法设置过滤查询语句
• 将配置好的源对象绑定到复制活动，完成任务编排

代码示例 #

import com.azure.resourcemanager.datafactory.models.*;
import java.util.List;

// 1. 定义源Kusto数据集（仅配置表和链接服务）
AzureDataExplorerTableDataset kustoSourceDataset = new AzureDataExplorerTableDataset()
    .withLinkedServiceName(new LinkedServiceReference().withReferenceName("你的源Kusto链接服务名"))
    .withTableName("Table");

// 2. 配置带过滤逻辑的Kusto源
AzureDataExplorerSource kustoSource = new AzureDataExplorerSource()
    .withQuery("Table | where $ingestionTime > ago(1d)"); // 直接传入过滤查询语句

// 3. 定义目标Kusto数据集（按需配置目标表和链接服务）
AzureDataExplorerTableDataset kustoSinkDataset = new AzureDataExplorerTableDataset()
    .withLinkedServiceName(new LinkedServiceReference().withReferenceName("你的目标Kusto链接服务名"))
    .withTableName("目标表名");

// 4. 构建复制活动
CopyActivity copyActivity = new CopyActivity()
    .withName("Kusto过滤复制任务")
    .withSource(kustoSource)
    .withSink(new AzureDataExplorerSink())
    .withInputs(List.of(new DatasetReference().withReferenceName(kustoSourceDataset.name())))
    .withOutputs(List.of(new DatasetReference().withReferenceName(kustoSinkDataset.name())));

// 5. 将活动加入管道（后续可发布到ADF实例）
Pipeline pipeline = new Pipeline()
    .withActivities(List.of(copyActivity));

关键说明 #

• withQuery()是AzureDataExplorerSource类的核心配置方法，直接传入符合Kusto语法的过滤语句即可
• 确认源表已启用$ingestionTime追踪（默认启用），否则查询会报错
• 提前确保链接服务拥有源集群的查询权限和目标集群的写入权限

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