使用流分析的 Azure IoT 中心云到设备消息

Question

我通过如下路径发送数据。

Android -> IoT Hub -> Stream Analytics -> SQL

我在流分析的查询中调用机器学习函数。现在，我想将机器学习的结果返回到Android设备。为了在 Android 设备上接收云到设备的消息，我已经使用 Device Explorer 进行了测试。我正在看官方教程，1 , 2但仍然不知道如何使用流分析将云发送到设备消息。 He表示使用服务总线和功能应用程序，但没有提供详细信息。我是 Azure 的新手。希望有人能给我一些指导或任何链接，以便我可以更多地了解如何实现它。提前致谢。

Answer 1

您可以使用Azure Function(预览版)输出 ASA 作业，以通过 Azure IoT 中心面向服务的端点发送云到设备消息。

以下是该函数的示例。

运行.csx:

#r "Newtonsoft.Json"

using System.Configuration;
using System.Text;
using System.Net;
using Microsoft.Azure.Devices;
using Newtonsoft.Json;

// create proxy
static Microsoft.Azure.Devices.ServiceClient client = ServiceClient.CreateFromConnectionString(ConfigurationManager.AppSettings["myIoTHub"]);

public static async Task<HttpResponseMessage> Run(string input, HttpRequestMessage req, TraceWriter log)
{
    log.Info($"ASA Job: {input}");

    var data = JsonConvert.DeserializeAnonymousType(input, new[] { new { xyz = "", IoTHub = new { ConnectionDeviceId = ""}}});

    if(!string.IsNullOrEmpty(data[0]?.IoTHub?.ConnectionDeviceId))
    {
        string deviceId = data[0].IoTHub.ConnectionDeviceId;
        log.Info($"Device: {deviceId}");

        // cloud-to-device message 
        var msg = JsonConvert.SerializeObject(new { temp = 20.5 });
        var c2dmsg = new Microsoft.Azure.Devices.Message(Encoding.ASCII.GetBytes(msg));

       // send AMQP message
       await client.SendAsync(deviceId, c2dmsg);
    }

   return req.CreateResponse(HttpStatusCode.NoContent);
}

函数.json:

{
  "bindings": [
    {
      "authLevel": "function",
      "name": "input",
      "type": "httpTrigger",
      "direction": "in"
    },
    {
      "name": "$return",
      "type": "http",
      "direction": "out"
    }
  ],
  "disabled": false
}

项目.json:

{
   "frameworks": {
   "net46":{
      "dependencies": {
        "Microsoft.Azure.Devices": "1.3.2"
      }
    }
   }
}

附录A

出于测试目的，请执行以下步骤:

1. 在 Azure Function 应用中添加设置 myIoTHub到 Azure IoT 中心的连接字符串。
2. 创建 HttpTrigger 函数，将其命名为 HttpASA_1
3. 使用上述内容更新 run.csx、function.json 和 project.json 文件
4. 使用以下示例进行测试请求正文:

[ { "时间": "2017-11-26T12:52:23.4292501Z", “柜台”:57， “风速”:8.7358， “温度”:16.63， “湿度”:79.42， "EventProcessedUtcTime": "2017-11-26T12:52:21.3568252Z", “分区ID”:2， "EventEnqueuedUtcTime": "2017-11-26T12:52:22.435Z", “物联网中心”:{ “消息ID”:空， “CorrelationId”:空， "ConnectionDeviceId": "设备1", "ConnectionDeviceGenerationId": "636189812948967054", "入队时间": "2017-11-26T12:52:21.562Z", “流ID”:空 } }]

将值 Device1 更改为您的实际 deviceId。

现在，AF 已准备好对其进行测试。按运行按钮运行示例并查看日志进度。

您应该会在设备上看到 AF 发送的 C2D 消息，或者您可以下载一个小型测试程序 Azure IoT Hub Tester模拟您的 MQTT 设备。以下屏幕片段显示了该测试仪:

现在，在这一步中，我们可以转到 ASA 作业来调用此 HttpASA_1 函数。请注意，ASA 作业仅调用 HttpTrigger 函数。以下屏幕片段显示了为我们的 Azure 函数添加输出:

当您在导入选项组合框中选择订阅时，您应该在组合框中看到此功能。完成后，按保存按钮并观看屏幕上的通知消息。 ASA 将向您的 AF 发送一条验证消息，其响应状态应为 20x 代码。

最后，您可以转到查询来为 AF 生成输出。以下屏幕显示了所有遥测数据到 AF 的简单输出:

注意，inpsim 是我的 iothub 输入。

ASA 按以下格式输出 HttpTrigger 函数的负载，请参阅我的示例:

[
  {
    "time": "2017-11-26T12:52:23.4292501Z",
    "counter": 57,
    "windSpeed": 8.7358,
    "temperature": 16.63,
    "humidity": 79.42,

    "EventProcessedUtcTime": "2017-11-26T12:52:21.3568252Z",
    "PartitionId": 2,
    "EventEnqueuedUtcTime": "2017-11-26T12:52:22.435Z",
    "IoTHub": {
      "MessageId": null,
      "CorrelationId": null,
      "ConnectionDeviceId": "Device1",
      "ConnectionDeviceGenerationId": "636189812948967054",
      "EnqueuedTime": "2017-11-26T12:52:21.562Z",
      "StreamId": null
    }
  }
]

请注意，我的遥测数据 (*) 是计数器、温度、湿度和时间戳时间，因此其他属性是由 ASA 作业隐式创建的。根据查询，您可以为 C2D 消息创建任何业务属性。