将数据从 Azure 流分析存储到 Blob 存储时，是否可以在路径中添加 DeviceID答案

【问题标题】：Can I add DeviceID in path while storing data from Azure Stream Analytics to Blob storage将数据从 Azure 流分析存储到 Blob 存储时，是否可以在路径中添加 DeviceID
【发布时间】：2019-07-17 06:35:17
【问题描述】：

我有从不同设备传入到 IoT 中心的数据，使用流分析对其进行处理并将其存储在 blob 存储中。我知道我们可以根据需要的格式在路径中添加{date}{time}，我们也可以在该路径中添加deviceId。

示例：对于 2018/10/30/01（日期/月/日/小时）可以在存储到 blob 时在该路径中添加 /deviceId

【问题讨论】：

此设备表示 2018/10/30/01/device1 2018/10/30/01/device2 我想将每个 deviceId 数据存储到该文件中
Not yet
不支持。作为一种解决方法，可以使用带有输出 blob 绑定的 HttpTrigger 函数。

标签： azure azure-storage azure-iot-hub azure-stream-analytics

【解决方案1】：

以下是针对您的情况的解决方法示例。它基于使用 azure 函数 (HttpTrigger) 输出 ASA 作业，以推送方式将数据附加到特定的 blob 存储。请注意，以下解决方法使用 Max batch count 将事件传递到 azure 函数值 1（当时一个遥测数据）。

ASA 职位查询：

SELECT
  System.Timestamp as [time], * 
INTO outAF
FROM 
  iot TIMESTAMP BY time

Azure 函数（HttpTrigger）：

运行.csx

#r "Newtonsoft.Json"
#r "Microsoft.WindowsAzure.Storage"

using System.Net;
using Microsoft.AspNetCore.Mvc;
using Microsoft.Extensions.Primitives;
using Microsoft.WindowsAzure.Storage.Blob;
using Newtonsoft.Json;
using Newtonsoft.Json.Linq;

public static async Task<IActionResult> Run(string body, CloudBlobContainer blobContainer, ILogger log)
{
    log.LogInformation($"{body}");

    var jtoken = JToken.Parse(body);
    var jobject = jtoken is JArray ? jtoken.SingleOrDefault<JToken>() : jtoken;
    if(jobject != null)
    {
        var jtext = jobject.ToString(Formatting.None);
        var data = JsonConvert.DeserializeAnonymousType(jtext, new {IoTHub = new { ConnectionDeviceId = ""}});        
        var blobName = $"{DateTime.UtcNow.ToString("yyyy/MM/dd/hh")}/{data.IoTHub.ConnectionDeviceId}";  
        var blob = blobContainer.GetAppendBlobReference(blobName);
        if(!await blob.ExistsAsync())
        {
            await blob.CreateOrReplaceAsync();
        }
        await blob.AppendTextAsync(jtext + "\r\n");
    }
return new NoContentResult();

}

函数.json

    {
      "bindings": [
       {
           "authLevel": "function",
           "name": "body",
           "type": "httpTrigger",
           "direction": "in",
           "methods": [
             "get",
             "post"
             ]
      },
      {
          "name": "blobContainer",
          "type": "blob",
          "path": "myContainer",
          "connection": "mySTORAGE",
          "direction": "out"
      },
      {
          "name": "$return",
          "type": "http",
          "direction": "out"
      }
      ]
 }

【讨论】：

谢谢。 Azure 函数的可扩展性是否足以让我们扩展到 10000 个设备每秒获取消息并触发此函数。你有它的节点 js 代码吗？
正如您在azure.github.io/azure-storage-node/… 中看到的那样，如果同时发生多个写入，则不确定是否会附加。在Options中我们不得不提到这个absorbConditionalErrorsOnRetry bool 指定是否吸收条件错误
请注意，使用函数可能会导致目标接收到的事件无序。如果您的业务逻辑取决于订单，您可能需要额外的逻辑来强制执行或解决这种情况（例如丢弃延迟的遥测）。

【解决方案2】：

我知道我们可以根据需要在路径中添加 {date}{time} 格式，在那个路径中我们也可以添加 deviceId。'

正如@Peter Bons 在comment 中提到的，目前不支持输出中的变量名。

作为解决方法，您可以使用Blob Trigger Azure Function。您需要在输出列中传递deviceId，然后在 blob 触发器函数中获取它。然后使用blob sdk创建/deviceId目录，将blob复制进去，删除之前的blob。

【讨论】：

我可以通过使用 blob SDK 来做到这一点。但是当我收到下一条消息时，它需要添加到那个 blob 中。例如，我添加了月/日/deviceId/abc.txt，当下次触发 azure 函数时它将替换 abc.txt 的内容，我无法附加到它。有没有办法附加到 blob ？特别是假设多个函数同时触发并且都试图追加到同一个 blob。
正如您在azure.github.io/azure-storage-node/… 中看到的那样，如果同时发生多个写入，则不确定是否会附加。在Options中我们不得不提到这个absorbConditionalErrorsOnRetry bool 指定是否吸收条件错误