如何通过.NET SDK在ADF复制活动中使用查询语句？

核心配置逻辑 #

复制活动的查询配置不需要修改已创建的源数据集本身，只需在复制活动的Source属性中指定对应数据源的源类，给对应的Query字段赋值即可，配置逻辑和门户UI完全对齐。

依赖安装 #

首先安装对应版本的NuGet包，示例采用官方最新的Azure资源管理SDK：

Install-Package Azure.ResourceManager.DataFactory

完整实现代码（以Azure SQL为源、Blob存储CSV为目标为例） #

using Azure.Core;
using Azure.Identity;
using Azure.ResourceManager;
using Azure.ResourceManager.DataFactory;
using Azure.ResourceManager.DataFactory.Models;

// 1. 认证并获取ADF实例资源
ArmClient armClient = new ArmClient(new DefaultAzureCredential());
ResourceIdentifier adfId = DataFactoryResource.CreateResourceIdentifier("你的订阅ID", "资源组名称", "ADF实例名称");
DataFactoryResource adfInstance = armClient.GetDataFactoryResource(adfId);

// 2. 配置源查询参数
AzureSqlSource sqlSource = new AzureSqlSource
{
    // 此处填入和门户UI完全一致的查询语句
    Query = "SELECT id, name, create_time FROM your_table WHERE is_deleted = 0",
    QueryTimeout = TimeSpan.FromMinutes(30)
};

// 3. 配置目标Sink（和你原有配置逻辑一致即可）
DelimitedTextSink blobSink = new DelimitedTextSink
{
    WriteBatchSize = 10000,
    WriteBatchTimeout = TimeSpan.FromMinutes(15)
};

// 4. 构造复制活动
CopyActivity copyActivity = new CopyActivity("查询复制活动")
{
    Source = sqlSource,
    Sink = blobSink,
    Inputs = { new DatasetReference("你提前创建的Azure SQL源数据集名称") },
    Outputs = { new DatasetReference("你提前创建的Blob存储目标数据集名称") }
};

// 5. 创建管道
DataFactoryPipelineData pipelineInfo = new DataFactoryPipelineData
{
    Activities = { copyActivity }
};
await adfInstance.GetDataFactoryPipelines().CreateOrUpdateAsync(WaitUntil.Completed, "你的管道名称", pipelineInfo);

不同数据源的配置说明 #

• 关系型数据库（MySQL/PostgreSQL/SQL Server等）：使用对应数据源的Source类，查询属性统一为Query，如果调用存储过程可替换为SqlReaderStoredProcedureName+StoredProcedureParameters属性
• 结构化文件（CSV/Parquet/ORC等）：使用对应格式的Source类，查询属性为Query，支持类SQL语法查询文件内容
• 动态参数场景：如需和门户UI一样通过参数动态传入查询内容，可将Query属性值设为@pipeline().parameters.自定义参数名，提前给管道配置对应参数即可

注意：源数据集不需要提前配置查询语句，活动级的Source配置优先级高于数据集配置，和门户UI的配置逻辑完全一致。

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