如何通过C#外部程序执行Salesforce Marketing Cloud的SQL查询

我希望自动化提取Salesforce Marketing Cloud（原Exact Target）中部分Data Extension的数据。目前我在Query Studio中运行如下SQL查询示例：

select distinct CompositeKey as ActivityCompositeKey, Type, BounceCategory from SendLogAnalyticsForExport where Type='EmailSent'

请问如何通过我将要编写的外部程序（优先使用C#）实现这一操作？

我拥有连接Marketing Cloud API的凭证，以及认证、REST和SOAP的URL：

AuthURL: https://XXXXXXXXXXX.auth.marketingcloudapis.com/v2/token
ClientId: YYYYYYYYYYYYYYYY
ClientSecret: ZZZZZZZZZZZZZ
SoapURL: https://XXXXXXXXXXX.soap.marketingcloudapis.com
RestURL: https://XXXXXXXXXXX.rest.marketingcloudapis.com

我查阅了「Marketing Cloud Engagement APIs and SDKs」文档，了解到REST API和SOAP API，但未找到针对上述SQL格式查询（即Query Studio中运行的类SQL查询）的相关内容。我找到最接近的是Salesforce「Data Cloud Reference Guide」中的「Query Data using Query API V1」文档，但它似乎并非针对Marketing Cloud，且说明十分简略：未提及实际调用的端点，也没有认证方法的相关信息。此外，我还查看了FuelSDK-CSharp库，但同样未找到针对Data Extension执行类SQL查询的相关内容。

在Marketing Cloud中，要执行类SQL查询（和Query Studio逻辑一致），需要通过Query Definition API创建、运行查询，再从生成的目标Data Extension中提取结果。以下是两种C#实现方式：

方式一：直接调用REST API实现 #

步骤1：获取API访问令牌 #

通过OAuth2认证获取访问令牌：

using System.Net.Http;
using System.Text;
using System.Text.Json;

public async Task<string> GetAuthToken()
{
    var authUrl = "https://XXXXXXXXXXX.auth.marketingcloudapis.com/v2/token";
    var clientId = "YYYYYYYYYYYYYYYY";
    var clientSecret = "ZZZZZZZZZZZZZ";

    var requestBody = new
    {
        client_id = clientId,
        client_secret = clientSecret,
        grant_type = "client_credentials"
    };

    using var client = new HttpClient();
    var content = new StringContent(JsonSerializer.Serialize(requestBody), Encoding.UTF8, "application/json");
    var response = await client.PostAsync(authUrl, content);
    response.EnsureSuccessStatusCode();

    var responseJson = await response.Content.ReadAsStringAsync();
    var tokenData = JsonSerializer.Deserialize<JsonElement>(responseJson);
    return tokenData.GetProperty("access_token").GetString();
}

步骤2：创建Query Definition #

提交SQL查询，指定用于存储结果的目标Data Extension：

public async Task<string> CreateQueryDefinition(string accessToken)
{
    var restUrl = "https://XXXXXXXXXXX.rest.marketingcloudapis.com/hub/v1/queryDefinitions";
    var querySql = "select distinct CompositeKey as ActivityCompositeKey, Type, BounceCategory from SendLogAnalyticsForExport where Type='EmailSent'";

    var queryDefinition = new
    {
        name = "Extract_SendLog_EmailSent", // 自定义查询名称
        description = "Extract EmailSent records from SendLogAnalyticsForExport",
        targetKey = "Target_DE_ExternalKey", // 目标DE的外部键
        queryText = querySql,
        categoryId = 12345, // 存储查询的文件夹ID，可在MC界面查询
        targetType = "DE"
    };

    using var client = new HttpClient();
    client.DefaultRequestHeaders.Authorization = new System.Net.Http.Headers.AuthenticationHeaderValue("Bearer", accessToken);
    var content = new StringContent(JsonSerializer.Serialize(queryDefinition), Encoding.UTF8, "application/json");
    var response = await client.PostAsync(restUrl, content);
    response.EnsureSuccessStatusCode();

    var responseJson = await response.Content.ReadAsStringAsync();
    var resultData = JsonSerializer.Deserialize<JsonElement>(responseJson);
    return resultData.GetProperty("id").GetString();
}

步骤3：运行Query Definition #

触发查询执行：

public async Task RunQueryDefinition(string accessToken, string queryDefinitionId)
{
    var restUrl = $"https://XXXXXXXXXXX.rest.marketingcloudapis.com/hub/v1/queryDefinitions/{queryDefinitionId}/actions/start";

    using var client = new HttpClient();
    client.DefaultRequestHeaders.Authorization = new System.Net.Http.Headers.AuthenticationHeaderValue("Bearer", accessToken);
    var response = await client.PostAsync(restUrl, null);
    response.EnsureSuccessStatusCode();
}

步骤4：从目标Data Extension提取结果 #

查询执行完成后，调用Data Extension API获取结果：

public async Task<string> GetDEData(string accessToken, string deExternalKey)
{
    var restUrl = $"https://XXXXXXXXXXX.rest.marketingcloudapis.com/data/v1/customobjectdata/key/{deExternalKey}/rowset?$pageSize=5000";

    using var client = new HttpClient();
    client.DefaultRequestHeaders.Authorization = new System.Net.Http.Headers.AuthenticationHeaderValue("Bearer", accessToken);
    var response = await client.GetAsync(restUrl);
    response.EnsureSuccessStatusCode();

    return await response.Content.ReadAsStringAsync();
}

方式二：使用FuelSDK-CSharp实现 #

FuelSDK-CSharp封装了SOAP API，可通过ETQueryDefinition类实现查询创建与执行：

using FuelSDK;
using System.Linq;

public void ExecuteQueryWithFuelSDK()
{
    var etClient = new ETClient
    {
        AuthStub = new ETSoapClient
        {
            Url = "https://XXXXXXXXXXX.soap.marketingcloudapis.com/Service.asmx",
            ClientId = "YYYYYYYYYYYYYYYY",
            ClientSecret = "ZZZZZZZZZZZZZ"
        }
    };

    // 创建Query Definition
    var queryDef = new ETQueryDefinition
    {
        Name = "Extract_SendLog_EmailSent",
        QueryText = "select distinct CompositeKey as ActivityCompositeKey, Type, BounceCategory from SendLogAnalyticsForExport where Type='EmailSent'",
        TargetType = "DE",
        TargetKey = "Target_DE_ExternalKey",
        CategoryID = 12345
    };

    var createResponse = queryDef.Post(etClient);
    if (!createResponse.Status)
    {
        throw new Exception($"创建查询失败：{string.Join(", ", createResponse.Results.Select(r => r.ErrorMessage))}");
    }

    // 运行查询
    var queryId = createResponse.Results.First().ObjectID;
    var runResponse = queryDef.Perform(etClient, "Start", new[] { queryId });
    if (!runResponse.Status)
    {
        throw new Exception($"运行查询失败：{string.Join(", ", runResponse.Results.Select(r => r.ErrorMessage))}");
    }

    // 轮询查询状态，确认执行完成（示例逻辑）
    bool isComplete = false;
    while (!isComplete)
    {
        var statusResponse = queryDef.Get(etClient, new[] { queryId });
        if (statusResponse.Status)
        {
            var status = statusResponse.Results.First().Status;
            isComplete = status == "Completed";
            if (!isComplete) System.Threading.Thread.Sleep(5000);
        }
        else
        {
            throw new Exception($"查询状态获取失败：{string.Join(", ", statusResponse.Results.Select(r => r.ErrorMessage))}");
        }
    }

    // 提取目标DE数据
    var de = new ETDataExtension
    {
        ExternalKey = "Target_DE_ExternalKey"
    };
    var retrieveResponse = de.Get(etClient);
    if (retrieveResponse.Status)
    {
        foreach (var result in retrieveResponse.Results)
        {
            // 读取字段值示例：var key = result.Properties["ActivityCompositeKey"]
        }
    }
}

关键说明 #

• 目标Data Extension需提前创建，字段结构要与SQL查询的输出完全匹配
• 查询执行是异步操作，需通过轮询Query Definition的状态确认执行完成（REST API可调用/hub/v1/queryDefinitions/{id}接口查询状态）
• 若只需简单过滤（无复杂SQL逻辑如distinct、多表关联），可直接调用Data Extension的Retrieve接口并指定过滤条件，无需创建Query Definition

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