如何在数据工厂中接收 http 帖子？

Question

函数 1 每 5 分钟通过 HTTP post 发送一个消息正文。 在数据工厂中，我想接收此消息并将此正文作为输入传递给函数 2。

{ “身体”：“跑” }

如何在 Azure 数据工厂中接收此消息？

Answer 1

Azure 数据工厂 (ADF) 支持 limited set of triggers。 http 触发器不是其中之一。

我建议让 Function1 直接调用 Function2。然后让 Function2 将数据存储在 blob 文件中。

之后就可以使用ADF的Storage event trigger来运行管道了：

存储事件触发器针对存储帐户中发生的事件运行管道，例如文件到达或 Azure Blob 存储帐户中文件的删除。

您可以从那里使用 ADF 管道获取数据并在管道中进一步处理。

Answer 2

好的，首先，您绝对可以通过 REST API 调用在 ADF 中触发管道执行，甚至可以将参数（例如正文内容）传递给该管道。

参考这个链接： https://docs.microsoft.com/en-us/rest/api/datafactory/pipelines/create-run

我现在不在电脑前，但明天我会举个例子。话虽如此，有很多方法可以做你想做的事，在 ADF 中和在 ADF 之外。

以下是从 HTTP 触发的 Azure 函数调用 Azure 数据工厂管道的示例。抱歉，我不熟悉 C#，所以代码可能看起来很糟糕，但我已经测试过了，它可以工作。

这是我用于 Azure 功能的代码：

#r "Newtonsoft.Json"

using System.Net;
using Microsoft.AspNetCore.Mvc;
using Microsoft.Extensions.Primitives;
using Newtonsoft.Json;
using System.Net.Http;
using System.Net.Http.Headers;
using System.Text;


public static async Task<IActionResult> Run(HttpRequest req, ILogger log)
{
    log.LogInformation("C# HTTP trigger function processed a request.");

    string name = req.Query["PipelineName"];

    string requestBody = await new StreamReader(req.Body).ReadToEndAsync();
    dynamic data = JsonConvert.DeserializeObject(requestBody);
    name = name ?? data?.PipelineName;

using(var client = new HttpClient())
{
//I Gave my Azure Function a system assigned Managed Identity, that way you can give RBAC roles for access to Management API and Azure Data Factory
//The below code gets the access token to be used in authenticating for the next API Call to ADF
client.DefaultRequestHeaders.Add("Secret", Environment.GetEnvironmentVariable("MSI_SECRET"));
var response = await client.GetAsync(String.Format("{0}/?resource={1}&api-version={2}", Environment.GetEnvironmentVariable("MSI_ENDPOINT"), "https://management.azure.com/", "2017-09-01"));
string msiResponse = await response.Content.ReadAsStringAsync().ConfigureAwait(false);
dynamic msiResponseData = JsonConvert.DeserializeObject(msiResponse);
//Access Token is saved below
string accessToken = msiResponseData?.access_token;
//return new OkObjectResult(accessToken);

using(var client1 = new HttpClient())
    {
    string uri = "https://management.azure.com/subscriptions/xxxxxxxxxxxxxxxxxx/resourceGroups/xxxxxxxxx/providers/Microsoft.DataFactory/factories/xxxxxxxxxxxxx/pipelines/ADFTestPipeline/createRun?api-version=2018-06-01";
    //string uri = "https://management.azure.com/subscriptions/[SubscriptionID]/resourceGroups/[ResourceGroupName]/providers/Microsoft.DataFactory/factories/[DataFactoryName]/pipelines/[PipelineName]/createRun?api-version=2018-06-01";
    string content = "{}";
        client1.DefaultRequestHeaders.Authorization = new AuthenticationHeaderValue("Bearer", accessToken);
        //The below Posts to the ADF API using the authorization header and bearer token, in order to pass parameters to the ADF Pipeline, you can modify the content string variable above (json format)
        var response1 = await client1.PostAsync(uri, new StringContent(content.ToString(), Encoding.UTF8, "application/json"));
        var result = await response1.Content.ReadAsStringAsync();
        return new OkObjectResult(result);


        //return new OkObjectResult(result);
    }
}
}

为了使其正常工作，您需要确保您的 Azure 函数具有系统分配的托管标识，然后有权访问您的 ADF 以运行管道，并有权调用 Azure 管理 API，这些可以两者都通过 RBAC 分配

设置系统分配的托管身份启用：

满足您的 AZ 功能需求：

• Azure 数据工厂贡献者
• API 管理服务贡献者（可能与 Operator 一起使用，尚未测试）

证明它有效：

Azure 函数输出/响应：

ADF 监视器选项卡：

我希望这对您有所帮助，并让您了解如何完成这项工作。如果它满足您的需求，请接受作为答案，如果您需要更多详细信息，请告诉我！