Извлечение информации из данных Интернета вещей с использованием потока данных холодного пути

Этот блог продолжает наше освещение руководства по решению, опубликованного командой Microsoft по отраслевому опыту. Руководство охватывает следующие компоненты:

• Прием данных
• Обработка по горячему пути
• Обработка по холодному пути
• Клиенты аналитики

Мы уже рассмотрели рекомендации по обработке данных для IoT-приложения в руководстве по решению и предложили использовать лямбда-архитектуру для потока данных. Повторим пути передачи данных:

• Пакетный уровень (холодный путь) хранит все входящие данные в их необработанном виде и выполняет пакетную обработку данных. Результат этой обработки сохраняется в виде пакетного представления. Это конвейер медленной обработки, выполняющий сложный анализ, объединяющий данные из нескольких источников в течение более длительного периода (например, часов или дней) и генерирующий новую информацию, такую как отчеты и модели машинного обучения.
• Уровень скорости и уровень обслуживания (теплый путь) анализирует данные в режиме реального времени. Этот уровень рассчитан на низкую задержку в ущерб точности. Это конвейер более быстрой обработки, который архивирует и отображает входящие сообщения, а также анализирует эти записи, генерируя краткосрочную критическую информацию и действия, такие как сигналы тревоги.

В этом сообщении в блоге рассматриваются компоненты холодной обработки в руководстве по решению.

Мы рассмотрели анализ временных рядов с помощью временных рядов Azure Информация (TSI) подробно описана в руководстве по решению. Это сервис аналитики, хранения и визуализации данных временных рядов. Пожалуйста, прочтите соответствующий раздел об использовании TSI.

Как вы, возможно, помните из предыдущих сообщений в блоге, мы используем выборочные данные, опубликованные конечной точкой NIST SMS Test Bed. Наши предыдущие публикации заканчивались тем, что данные были перенесены в отдельные концентраторы событий Azure для записей “события” и “образцы” данных.

“#xD; “#xD;

Прежде чем мы начнем остальную часть обсуждения, мы хотели бы подчеркнуть, что решение проблемы “аналитики” зависит от каждого завода, линии, станка и так далее. Данные должны быть доступны и соответствовать потребностям бизнеса. Мы рассмотрим два разных подхода к организации данных, но они не являются исчерпывающими и предназначены только для примеров.

Хранение необработанных данных

Наш образец реализация содержит базовый набор запросов Azure Stream Analytics, который принимает входящий поток данных из концентраторов событий, в которые отправляются необработанные данные, и копирует его в большие двоичные объекты и таблицы хранилища Azure. В качестве примера запросы выглядят следующим образом:

ВЫБЕРИТЕ * В [samplesTable] ИЗ [EventHubIn]

Одна таблица предназначена для выборок, а другая – для событий. Поскольку мы сглаживали входящие данные в пользовательском компоненте, мы добавили свойство для часового окна, в котором находился поток входящих данных, используя следующий фрагмент кода на C#, чтобы упростить организацию данных в конвейерах обработки:

часовое окно = новая дата и время( образец.временная метка.Год, образец.временная метка.Месяц, образец.временная метка.День, образец.временная метка.Час, 0, 0),

Это поле записи данных особенно полезно при организации записей в таблице хранилища Azure, просто используя его в качестве ключа раздела. Мы используем порядковый номер входящей записи в качестве ключа строки. Объектная модель для таблиц хранилища описана в документации “Понимание модели данных службы таблиц”. Пожалуйста, также ознакомьтесь с документацией “Разработка масштабируемой стратегии разделения для хранилища таблиц Azure” для получения рекомендаций по проектированию таблицы хранилища.

Большие двоичные объекты хранилища Azure, сгенерированные заданием ASA, организованы в контейнеры для каждого часа в виде одного большого двоичного объекта для данных за час в формате значений, разделенных запятыми (CSV). Мы будем использовать их в будущем для нужд искусственного интеллекта (ИИ).

Загрузка данных в базу данных SQL Azure

Мы рассмотрим базовый способ постепенно загружайте записи в базу данных SQL Azure, а затем обсудите возможные способы их дальнейшей обработки для создания новых агрегатов и сводных данных.

Наша цель – предоставить простой подход, чтобы показать, как данные могут поступать в хранилища данных, и продемонстрировать технологии, полезные для этого. Любое аналитическое решение в значительной степени зависит от контекста и требований, но мы попытаемся предоставить базовые механизмы для демонстрации соответствующих служб Azure.

Azure Data Factory (ADF) – это служба облачной интеграции для создания служб хранения, перемещения и обработки данных в автоматизированных конвейерах передачи данных. У нас есть простой конвейер ADF, который демонстрирует постепенную загрузку таблицы, используя таблицу хранилища в качестве источника.

В конвейере есть действие поиска, которое выполняет следующий запрос к базе данных SQL:

выберите ПРЕОБРАЗОВАТЬ( символ(30), случай, когда max(SampleTimestamp) равно null, тогда ‘1/1/2010 12:00:00 AM’ else max(SampleTimestamp) end, 126) как LastLoad из [Samples]

Стиль, используемый в функции ПРЕОБРАЗОВАНИЯ, 126, обозначает значение временной метки, которое должно быть отформатировано как “гггг-мм-ДДТЧ:mi:ss.mmm,” который соответствует строковому представлению значения ключа раздела в таблице хранения. Запрос возвращает последнюю запись, которая была перенесена в базу данных SQL. Затем мы можем передать это значение следующему действию, чтобы запросить хранилище таблиц для извлечения новых записей.

Далее выполняется действие “Копировать данные”, которое просто использует возвращаемое значение из действия поиска, которое является значением “Последней загрузки”, и выполняет следующий запрос таблицы для источник. Пожалуйста, обратитесь к запросам таблиц и сущностей для получения подробной информации о запросах таблиц хранилища.

SampleTimestamp gt datetime’@{FormatDateTime(activity(‘LookupSamples’).вывод.firstRow.LastLoad, ‘гггг-ММ-ДДТХ:мм:сс.fffZ’)}’&# xD;

Позже это действие сопоставит столбцы таблицы хранилища (свойства) со столбцами таблицы базы данных SQL. Этот конвейер планируется запускать каждые 15 минут, таким образом, постепенно загружая целевую таблицу базы данных SQL.

Примеры обработки

Дальнейшая обработка необработанных данных зависит от фактического требования. В этом разделе рассматриваются два потенциальных подхода к обработке и систематизации данных для демонстрации возможностей.

Давайте сначала рассмотрим данные, которые мы собираем, чтобы узнать подробности. Обратите внимание, что исходные данные в таблице samples представлены в виде пар имя/значение. Первый запрос предоставит нам различные типы выборок, записанные каждой машиной.

ВЫБЕРИТЕ DeviceName, ComponentName, SampleName, COUNT(SampleSequence) В КАЧЕСТВЕ sampleCount ИЗ образцов ГРУППИРУЙТЕ ПО DeviceName, ComponentName, SampleName УПОРЯДОЧИВАЙТЕ ПО DeviceName ASC, ComponentName ASC, SampleName ASC, sampleCount DESC

Мы наблюдаем, что существует восемь машин, и каждая из них отправляет разные наборы типов выборок. Ниже приведен частичный результат предыдущего запроса. Мы немного подробнее проанализировали результат в Microsoft Excel, чтобы дать представление об относительном количестве выборок:

Мы можем заключить, что лучший способ агрегировать и суммировать результаты – это сначала организовать результаты по машинам – например, таблица необработанных данных для каждой машины.

Здесь мы будем шаг за шагом демонстрировать концепции. Некоторые читатели, несомненно, найдут более оптимизированные способы реализации некоторых запросов, но наша цель здесь – привести четкие примеры, демонстрирующие концепции.

Возможно, мы пожелаем дополнительно обработать данные, сначала переместив необработанные данные, которые находятся в пары имя/значение следующим образом:

Мы можем использовать следующий запрос для создания новой таблицы и переноса целых строк. Этот запрос предполагает, что мы не различаем ни один из компонентов и видим машину в целом:

С machine08samples, отображаемым КАК ( ВЫБЕРИТЕ * ИЗ ( ВЫБЕРИТЕ SampleTimestamp, sampleName, ПРИВЕДИТЕ(sampleValue КАК ЧИСЛОВОЕ(20,3)) В КАЧЕСТВЕ sampleValueNumeric ИЗ Samples ГДЕ DeviceName = ‘Machine08’ и ISNUMERIC(sampleValue) != 0 ) В КАЧЕСТВЕ S PIVOT( MAX(sampleValueNumeric) ДЛЯ имени выборки В ([S2temp], [Stemp], [Zabs], [Zfrt], [S2load], [Cfrt], [total_time], [Xabs], [Xload], [Fact], [Cload], [cut_time], [Zload], [S2rpm], [Srpm], [auto_time], [Cdeg], [Xfrt], [S1load]) ) В ВИДЕ сводной таблицы ) SELECT * INTO Machine08Samples FROM Machine08SamplesTransposed

Мы можем перенести этот запрос в конвейер ADF, переместив его в хранимую процедуру с параметром для запроса необработанной таблицы, чтобы только вводятся последние загруженные строки, и изменяем “SELECT * INTO” на “INSERT * INTO”. Мы рекомендуем максимально полагаться на хранимые процедуры, чтобы эффективно использовать ресурсы базы данных SQL.

Результирующая таблица выглядит следующим образом (некоторые столбцы удалены для краткости).

Одним из способов обработки этого промежуточного набора данных является заполнение нулевыми значениями выборок из последнего полученного значение, как показано ниже.

Мы должны подчеркнуть, что мы не рекомендуем это решение для каждого бизнес-кейса и для каждого выборочного значения. Такой подход имеет смысл для значений, которые значимы вместе. Например, в определенном случае группировка Fact (фактическая скорость подачи по пути) и Zfrt (скорость подачи по оси Z) может иметь смысл. Однако для другого случая Xabs (абсолютное положение по оси X) и Zfrt в одной записи, сгруппированные таким образом, могут не иметь смысла. Группировка значений выборки должна выполняться в каждом конкретном случае, в зависимости от потребностей бизнеса.

Или другой способ – поместить отдельные записи в временные сегменты и применить агрегатную функцию в этой группе:

Давайте приведем небольшой пример для достижения первого вариант. В предыдущем примере мы получили V1.1 в момент t1 и получили V2.2 в момент t2. Мы хотим заполнить значение Sample1 для t2 с помощью t1s, V1.1.

С NonNullRank КАК ( ВЫБЕРИТЕ SampleTimestamp, S2temp, cnt = COUNT(s2temp) OVER (ПОРЯДОК ПО SampleTimestamp) ИЗ Machine08Samples ), WindowsWithNoValues КАК ( ВЫБЕРИТЕ SampleTimestamp, S2temp, r = ROW_NUMBER() ПОВЕРХ (РАЗБИЕНИЕ ПО ПОРЯДКУ cnt ПО SampleTimestamp ASC) – 1 ИЗ NonNullRank ) ВЫБЕРИТЕ SampleTimestamp, S2temp, S2tempWithValues= ISNULL(S2temp, LAG(S2temp, r) OVER (ПОРЯДОК ПО SampleTimestamp ASC)) ИЗ WindowsWithNoValues

Когда мы анализируем предыдущие запросы, первое общее табличное выражение (CTE), NonNullRank, дает нам ранг ненулевых значений выборочных значений S2temp среди полученных записей данных.

Второй CTE, WindowsWithNoValues, предоставляет нам окна образцов с полученное значение вверху и порядок нулевых значений внутри окон (столбец r).

Заключительный запрос заполняет нулевые значения с помощью аналитической функции запаздывания, перенося полученное значение из верхней части окна в текущую строку.

Второй вариант, о котором мы упоминали ранее, заключается в группировании полученных значений и применении агрегатной функции внутри группы.

С С 30-секундными пакетами В виде ( SELECT *, (dateadd(second,(datediff (second,’2010-1-1′,[ SampleTimestamp])/(30))*(30),’2010-1-1′)) КАК [SampleTimestamp30Seconds] ИЗ Machine08Samples ) ВЫБЕРИТЕ SampleTimestamp30Seconds, AVG(S2Temp) ИЗ With30SecondBuckets ГРУППИРУЙТЕ ПО SampleTimestamp30Seconds УПОРЯДОЧИВАЙТЕ ПО sampletimestamp30секунд

Мы можем поместить эти запросы в хранимую процедуру для создания новых агрегированных и сводных таблиц по мере необходимости для использования аналитическим решением.

Здесь мы хотели бы еще раз повторить наш вступительный аргумент. Решение проблемы аналитики зависит от имеющихся данных и потребностей бизнеса. Возможно, не существует какого-то одного решения, но Azure предоставляет множество технологических опций для реализации данного решения.

Следующие шаги

• Завершите учебное пособие по ADF для преобразования данных в облаке с помощью действия Spark и по требованию Другой пример – служба, связанная с Azure HDInsight.
• Получите более полную картину извлечения информации из данных Интернета вещей из руководства по решению.