Визуализация данных с IoT Edge устройств. Stream Analitycs Job Query. Урок 8.

Для визуализации данных полученных с термо сенсора в предыдущей статье добавим их сохранение в CosmosDB.

1. На Azure Portal в Create Resources -> Databases -> Azure Cosmos DB -> создаем базу данных и коллекцию «TermoSensorDBCollection».
2. На Azure Portal в Create Resources -> Internet of Things ->  Stream Analytics Job:

   JobName - "TermoSensorStreamAnalytics"
   Subscription - Free Trial
   Resource Group - [у меня - SchneiderEq]
   Location - [West Europe]

   Жмем кнопку «Create».

3. Переходим в All Resources -> TermoSensorStreamAnalytics (созданный Stream Analytics Job).
4. В разделе Job topology -> Inputs -> Add stream input -> IoT Hub:

   Input alias: TermoSensorStreamAnalyticsInput
   Выбираем: Select IoT Hub from your subscription
   Subscription: Free Trial
   IoT Hub: [мой IoT Hub SchneiderEq]
   Endpoint:Messaging
   Consumer group: $Default (оставляем по умолчанию)
   Event serialization format: JSON
   Encoding: UTF-8
   Event compression type: None

   Нажимаем «Save».

5. В разделе Job topology -> Outputs -> Add -> Cosmos DB:

   Output alias: TermoSensorStreamAnalyticsOutput
   Выбрать: Select Cosmos DB from your subscriptions
   Subscriptions: Free trial
   Database: Use Existing (предполагается, что вы уже создали Cosmos DB).
   Collection name pattern: TermoSensorDBCollection

   Нажимаем «Save».

6. Теперь очень важный момент, на который нужно обратить пристальное внимание. В разделе Job topology -> Query нужно написать корректный запрос. Нюанс состоит в том, что простой запрос вида:

   SELECT
       *
   INTO
       TermoSensorStreamAnalyticsOutput
   FROM
       TermoSensorStreamAnalyticsInput

   свалит в CosmosDB в одну кучу информацию с различных устройств передающих данные в IoT Hub. В дальнейшем это усложнит анализ в PowerBI, поскольку поля в JSON будут разные. Можно разделить записи по разным таблицам используя SELECT, но это менее удобно.

   Так что такой простой скрипт использовать не стоит!

   Добавим в запрос WHERE, чтобы исключить данные от других устройств. Для этого посмотрим с помощью консольной утилиты, какие служебные поля передает IoT Edge устройство в IoT Hub и напишем запрос с фильтрацией по конкретному устройству.В случае с Modbus устройством в IoT Hub передается параметр HwId с именем устройства заданном в properties.desired модуля modbus. Можно фильтровать по этому полю, либо по имени самого устройства в IoT Hub. В последнем случае запрос будет выглядеть следующим образом:

   SELECT
       *
   INTO
       TermoSensorStreamAnalyticsOutput
   FROM
       TermoSensorStreamAnalyticsInput
   WHERE TermoSensorStreamAnalyticsInput.IoTHub.ConnectionDeviceId = 'TermoSensorWithABB'

   В этом случае в TermoSensorStreamAnalyticsOutput будут направляться только данные от  «TermoSensorWithABB». Протестировать работу запроса можно, подав на INPUT сохраненный тестовый файл с перехваченной JSON последовательностью.
   Не забываем нажать «Save» для сохранения запроса.

7. Если нужно записывать телеметрические данные поступающие с разных устройств в соответствующие collections одной базы CosmosDB, то на каждую collection создается отдельный Stream Analytics Job со своим QUERY  в котором в WHERE стоит правильная фильтрация.
8. После настройки не забываем запустить Stream Analytics Job, иначе поступающие данные будут уходить в null.  
9. После запуска сервиса iotege на устройстве IoT Edge начинают поступать данные на Stream Analytics Job и передаваться в CosmosDB.
10. Далее CosmosDB подключается к PowerBI. На эту тему есть достаточно подробная статья, поэтому нет смысла рассматривать этот момент отдельно.

Другой способ сохранить данные в CosmosDB — использовать Azure Functions, но это более трудоемкий процесс. Он подробно описан в статье.

И ещё один вариант через маршрутизацию сообщений описан в следующей статье.