azure - 发送有关在 azure data Lake gen 1 中创建新文件的事件

Question

将新文件添加到第 1 代 Azure Data Lake 后，我希望向外部 NIFI 流发送事件或通知。

有人工作过或有关于此用例的任何信息吗？

Answer 1

嘿，这是一个非常好的问题，不幸的是，目前没有可以连接的开箱即用事件，但我找到了一种使用诊断日志来做到这一点的方法。

所以这个想法很简单

1. 在 ADLSv1 上设置诊断日志，以将日志流式传输到事件中心
2. 为事件中心上的新整体触发逻辑应用

<小时/>

因此要在 ADLSv1 上设置诊断日志

1. 转到 ADLSv1 的诊断设置

添加诊断设置

设置指向您的事件中心的链接

<小时/>

现在设置逻辑应用

1. 打开逻辑应用设计器

添加事件中心触发器(假设存在名为 stackdemohub 的中心)

添加过滤器数组操作

3.1。将 From 设置为表达式

triggerBody()?['ContentData']['records']

3.2。将右侧条件设置为“创建”，将左侧条件字段设置为表达式

item()['operationName']

添加 for-each 循环并从过滤器数组步骤传递正文

保存并运行

5.1。当逻辑应用成功执行时您将看到

您将在列表中找到新文件

如您所见，我上传的文件之一在演示文件夹中名为 MarketplaceCharges.json。

每个事件都是这样的

{
    "time": "2019-09-18T07:48:20.342Z",
    "resourceId": "/SUBSCRIPTIONS/2BCB9F3D-3F6B-4345-A49E-86D3141C7F73/RESOURCEGROUPS/STACKDEMO/PROVIDERS/MICROSOFT.DATALAKESTORE/ACCOUNTS/STACKDEMO",
    "category": "Requests",
    "operationName": "create",
    "resultType": "201",
    "callerIpAddress": "::ffff:111.222.333.444",
    "correlationId": "93faafd5-dfa2-4432-91f8-c7f360d80655",
    "identity": "<a href="https://stackoverflow.com/cdn-cgi/l/email-protection" class="__cf_email__" data-cfemail="08696c69654865697a6b726963266167" rel="noreferrer noopener nofollow">[email protected]</a>",
    "properties": {
      "HttpMethod": "PUT",
      "Path": "/webhdfs/v1/demo/MarketplaceCharges.json",
      "RequestContentLength": 0,
      "ClientRequestId": "288c654f-0948-4468-8e92-b158cc265c54",
      "StartTime": "2019-09-18T07:48:20.264Z",
      "EndTime": "2019-09-18T07:48:20.334Z",
      "UserId": "8162E212-E32B-443C-8F13-1CDA7B264DDB"
    }
}

并且您可以通过文件路径 /webhdfs/v1/demo/MarketplaceCharges.json

获得值(value)

我在 ADLSv1 上创建了 3 个文件，并按预期在循环中获得了 3 个项目

现在您可以对此事件信息执行任何您想要的操作，并将其发送到您需要的任何地方。

<小时/>

在结束语中，您可能希望将逻辑应用程序更改为函数应用程序，以防请求量较大，因为大规模逻辑应用程序并不便宜

[FunctionName("EventHubTriggerCSharp")]
public static void Run([EventHubTrigger("stackdemo", Connection = "EventHubConnectionAppSetting")] EventData[] eventHubMessages, ILogger log)
{
    foreach (var message in eventHubMessages)
    {
        // parse items and do something
    }
}

对于较大的文件，还有一个重要说明，请确保为进程添加一些延迟，因为创建的请求类型是当文件出现在存储上时，即使它仍在复制。在这种情况下，将会有更多的“追加”事件发生。