Stream Analytic - обновить строку в таблице, не добавить данные

Question

Всем привет. 

Есть ли какой-нибудь механизм, чтобы с помощью SA обновить данные в таблице, не добавить новые строки? 

Есть таблица tlmt - там хранятся исторические данные. Есть таблица current - в ней текущие. 

Можно ли в current обновлять строку для соответствующего устройства, а в tlmt также оставлять историю? 

Обе таблицы в базе данных, которая подключена к Mobile Services. 

Заранее спасибо. 

Answer 1

Как мне известно, все выходы в Stream Analytics работают только на добавление. Это полностью соответсвует идее большинства сервисов в Azure, Stream Analytics решает только одну задачу обработки потока событий, но делает это хорошо.

Так как обе таблицы лежат в БД, то решал бы задачу я следующим образом. Отправил бы исходящий поток для подсчета таблицы current со всей необходимой информцией в новый Event Hub (или, еще лучше, Azure Storage Queue или Service Bus Queue). А полчающийся поток сообщений обработал с помощью Web Jobs или Azure Functions и отправил в БД в форме обновления (на C# или JavaScript). Оба имеют очень удобную интеграцию с Event Hub и очередьми, а потому решение едва ли будет больше пары строк.

То есть, схема такая.

Источник данных телеметрии -> Stream Analytics -> Storage Queue -> Azure Functions -> Azure SQL Database

Исторический же поток есть смысл оставить как есть.

Answer 2

Спасибо, буду копать. Отпишусь по результатам.

Answer 3

Ковыраю третий час, так и не понял как информацию отправить в Storage Queue из SA. 

1. Создал Storage Account. 

2. В SA добавил output: 

Никаких ошибок не было замечено ни с какой стороны. Но данных нет. Не подскажете, что делаю не так? Или нужно было выбрать Blob Storage?

Answer 4

Покажите запрос в Query?

Answer 5

Забыл указать INTO "новый output". 

Данные пишутся, спасибо. 

Вопрос по Azure Functions - как там можно писать в Azure SQL? \

Вот аутпуты, доступные там:

Answer 6

Вынесите, пожалуйста, этот вопрос в новый вопрос и пометьте данный выбором правильного ответа (если он был), чтобы будущим пользователям было удобнее находить вопрос. :) 

Касаемо вашего вопроса про SQL - согласно документации, SQL пока не поддерживается. Вы можете же в SQL писать сразу из Stream Analytics - создайте соответствующую схему и таблицу в SQL Azure, и укажите БД в качестве output.

Answer 7

Извините, если покажется грубым - почитайте моё первое сообщение и почему я так не могу сделать:) Мне нужно обновить строку в таблице, не добавить. evgri243 посоветовал такую схему с SA -> Blob (или Table) -> Functions -> SQL. 

А теперь оказывается в SQL не могу писать через функции) 

P.S: ответом пометить не могу, т.к. проблему он не решает, извините.

Answer 8

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

Предполагаю, что я не самым ясным образом выразился. Да, такого output'а нет, да, если он и появится, то опять же будет скорее всего только на добавление (как и все приведенные). Ваш паттерн использования выглядит достаточно стандартным, но его не так легко привести под общий знаменатель, работающий под всех. Под Azure Functions и Web Jobs я имел в виду некоторую среду исполнения, которая сможет выполнить произвольную логику обновления на C# или JS с требуемыми обработками исключительных случаев. В данном конкретном случае я бы написал на базе Entity Framework или ADO .NET (скорее на ADO.NET) небольшой код, отправляющий в сторону базы выражение MERGE. Это позволит за одно обращение и обновить существующие данные или добавить новые, в случае их отсутствия.

О выборе между сервисами. Web Jobs и Functions по сути одно и тоже и отличаются они лишь моделью хостинга. В случае с WebJobs вы разрабатываете проект в Visual Studio и потом хостите его в App Service, в случае Functions -- пишите код на сайте и отдаете часть вопросов хостинга на откуп Microsoft, платя за это более серьезными ограничениями и отсутствие полноценного Debugging (особенно с учетом раннего Preview, в дальнейшем ситуация должна значительно улучшиться). Functions скорее всего хватит для ADO .NET, в то время как для EF я бы сейчас предпочел использовать WebJobs.

Немного о моей логике за выбором именно такой архитектуры. WebJobs и Functions поддерживают автоматическое горизонтальное шкалирование, это позволит в случае получения большого потока сообщений со стороны Stream Anaytics (а иначе зачем нужен Stream?) справиться с нагрузкой. Очередь, в данном случае реализующая паттерн Service Bus, помогает обеспечить это шкалирование, обрабатывая эти сообщения по порядку и без опасности нарваться на различные формы переполнений. Например, они могут возникнуть при прямом обращении к какому-нибудь своему REST-сервису (переполнение HTTP очереди в IIS) или переполнение памяти в "узком горлышке" самого сервиса. Скажу честно, сталкивался с последним и теперь отдаю предпочтение Service Bus при работе с потоками данных любой природы.

Answer 9

Да, это вариант. WebJob, работающий как сервис, в данном случае выглядит самым приемлемым решением. То есть WebJob достает сообщения из Event Hub, формирует запрос и отправляет на MERGE, вызывая API мобильного сервиса. Вообще, я бы здесь попробовал абстрагироваться от самой базы данных и выставить наружу Custom API для обновления записи и делать это руками мобильного сервиса, а не обращаться напрямую в БД. 

Answer 10

Alex, спасибо за ответ. Custom API боюсь выставлять наружу по двум причинам:

1. Можно будет отправить запрос на него с любого компьютера, тем самым инициировав обновление данных в таблице. Т.е. просто отправив правильный POST/GET запрос записать в таблицу фэйковые данные. По крайней мере, я не нашел варианта, как создать подключение к API используя ключ через Azure Functions/WebJobs. 

2. Я в принципе могу Custom API выставить наружу и конечные устройства путем выполнения запроса будут обновлять данные в таблице current. Но во-первых, это как-то не спортивно. Во-вторых, когда будет устройств около 1000, я боюсь, что сервер не даст выполнять сразу 1000 экземпляров процесса nodejs (если не прав - поправьте). 

Answer 11

evgri243, спасибо за столь подробный ответ. Примерно понял разницу между Web Jobs и Functions, но всё равно нужно больше мат. части. Буду читать. 

По вашей модели - как предполагается подключение к базе данных? Через Custom API я бы не хотел это делать (ответил выше пользователю Alex). Или вы имеете в виду подключение к SQL-серверу "напрямую"? Хотел бы этот момент прояснить. 

Теперь у меня схема такая:

Device -> IoT Hub -> Stream Analytic -> SQL (история)

                                                            |->Blob Storage (буфер) -> Azure Functions (обработка Blob Storage) -> ?

Эта схема только в голове. Пока ещё не понял, каким именно способом читать Blob Storage или Table Storage - запросом брать последние данные? Если да - то не будет ли увеличиваться время выполнения запроса с ростом записей? 

P.S: на данный момент я "выкрутился" другим способом. Написал в scheduler скрипт на nodejs, который берёт последние данные с таблицы tlmt и обновляет запись в current для каждого устройства. Сейчас в базе 4 устройства, скрипт отрабатывает моментально. Не знаю, насколько быстро будет выполняться при большем количестве устройств. 

Может кому-то будет полезно, кусок скрипта (я в nodejs не силен, поэтому если есть какие-то коммментарии - рад буду услышать):

function updateCurrentForAllDevices(){    
    deviceTable.select('deviceId')
        .read({ success: function(results) {
                            var devices = results;
                            devices.forEach(findCurrentEntity);    
                        }
               });
}


//найти id устройства в таблице current  
function findCurrentEntity(element, index, array)
{
    var deviceId = element.deviceId;
    var curId = -1;    
    //получить id строки в current
    currentTable        
        .where({deviceId: deviceId})
        .read({
            success: function(result){               
                    var cur = result;                
                    curId = cur[0].id;
                    var tlmtId = cur[0].tlmtId; 
                    var pushCount = cur[0].pushCount;                                       
                    getTlmtEntity(deviceId, curId, tlmtId, pushCount);
                }
               })
}

//Получить последнюю строку телеметрии данного устройства в таблице current
function getTlmtEntity(deviceId, curId){    
    tlmtTable
        .orderByDescending('id')                        
        .where({deviceId: deviceId})                        
        .take(1)                        
        .read({ 
            success: function(results) {                            
                            var tlmtEntity = JSON.stringify(results);
                            //console.log('current tlmtId = ' + tlmtId + ' tlmt id = ' + results[0].id + ' pushCount = ' + pushCount);                                            
                            updateCurrentTlmt(curId, tlmtEntity);                                                                                       
                        }
               });  
}


//Обновить таблицу current
function updateCurrentTlmt(curId, tlmtEntity, tlmtId, pushCount)
{
    var tlmt = JSON.parse(tlmtEntity);        
       
         
    currentTable.update(
    {
        id:         curId,
        deviceId:   tlmt[0].deviceId,
        light:      tlmt[0].light,
        temp:       tlmt[0].temp,
        hmdt:       tlmt[0].hmdt,
    })

    
}

Answer 12

Прошу прощения за очень длинный ответ, но я решил высказаться, ибо подобный вопрос с использованием Stream Analytics уже не в первый раз возникает и мне хотелось бы максимально подробно изложить свою точку зрения на этот счет.

Так, давайте по порядку. Идея с сервисом может быть очень актуальной, если помимо просто записи данных вы хотите что-то сделать при любых изменениях. Это позволяет абстрагировать базу за дополнительным уровнем и иметь единственную точку входа в сервис со своей логикой. Если такой нужды пока нет, то можно напрямую писать в базу.

Позвольте начать немного с самого начала, чтобы прояснить вопрос с архитектурной точки зрения до конца. У вас, к примеру, в будущем ожидается десятки тысяч-миллионов устройств, отсылающие телеметрию раз в секунду. Если писать эти данные напрямую в базу, то база содрогнётся от того, что помимо вставки ей еще и индексы надо обновить и кучу ограничений целостности проверять (банковские системы делают это, но они и стоят дорого и устроены сложно). В нашем случае нужен буфер, который гарантированно примет огромный поток данных, а уже потом позволит их не спеша обработать. Этот Event Hub. Так, мы научились гарантировать, что данные хотя бы дойдут до нас, а не будут сброшены в результате перегруза.

Но Event Hub -- это очень хороший буфер, но все-таки просто буфер -- с ним особо ничего кроме этого не сделаешь. Часто паттерн использования данных такой, что для оперативной (читай, близкой к реальному времени) обработки, деятельности организации, обслуживания пользователей или еще чего нужен не весь поток, а, например, каждое 10-е сообщение или средние срезы по часовым интервалам, да еще и с хитрой фильтрацией. А результаты обработки подробных данных пользователь может и подождать в случае необходимости. Это то где на сцену выходит Stream Analytics. Подробный поток вы можете отправить на относительно дешевый Storage для дальнейшей обработки Hadoop-подобными решениями по запросу (и нечто подобное вами уже заложено в идее таблицы tlmt). Данные же для оперативной обработки (ваша таблица current) могут на уровне Stream Analytics проходить дополнительную аггрегацию или фильтрацию, значительно сокращая поток исходящих в сторону базы событий. Итак, теперь у нас два потока -- один с подробными историческими данными для обработки по запросу и несильно огромный поток для оперативной работы.

Подробные данные пусть складируются, их оптимальное хранение и обработка -- тема совсем другого разговора. Вернемся к оперативным. Скорее всего у нас не просто процесс вставки данных, а что-то более сложное продиктованное бизнес требованиями (может нам надо оповещать пользователя, если значение стало слишком большим или вообще делать сложную аналитику на чем-то близком к машинному обучению). На практике все это может занимать некоторое время, что на уровне работы с потоком напрямую породило бы узкое горлышко, которое опять бы для нас окончилось либо сбросом данных или Memory Overflow, котрый бы ронял все решение безоговорочно. Какие нам нужны требования к решению на этом шаге:

1. Логика обработки для оперативной работы может в теории быть очень сложной, а потому интерфейс должен быть максимально гибким для создания собственной сложной логики.
2. Готовность к тому, что в одни моменты времени запросов на обработку будут сотни тысяч, а в другие -- просто сотни. Это не должно создавать проблем с перевыделением ресурсов (сотни машин для обработки 2 запросов в час) или их недостатком (2 машины для обработки сотни тысяч запросов в минуту). Решение должно шкалироваться и желательно делать это автоматически.
3. Запросы должны обрабатываться параллельно, а не последовательно (следствие из предыдущего пункта). Тут одно из фундаментальных ограничений всей идеи -- обработка должны быть порядко-независимой или, как минимум, неверный порядок должен отлавливаться и обрабатываться особым образом.
4. Должна быть гарантия доставки запроса до обработчика (пусть даже это и займет время в моменты высокой загрузки). Ибо это очень плохо, если из-за проблем с ресурсами мы начнем просто терять данные.
5. Мы не боги, а потому в случае ошибки обработки (база резко стала недоступна или запрос на вставку был признан Deadlock-жертвой и прибит) этот факт должен логироваться, а запрос отправляться на повторную обработку. И повторы должны производиться до определенного уровня, чтобы не создавать каруселей сообщений, которые и делают, что постоянно падают.

Все эти требования описывают один из очень стандартных паттернов. Идея безумно проста и выполняет роль фундаментальной в том, что сейчас принято называть "облачной архитектурой". Создать очередь, в которую публикуются сообщения, и подключить к этой очереди обработчиков, которые по мере освобождения берут из нее новое сообщение и обрабатывают его в своем ритме и со своими ресурсами.

Что делает очередь? Это точка, куда другими приложениями публикуются сообщения для последующей обработки. Очередь чаще всего в контексте этого паттерна в той или иной степени сохраняется на жесткий диск, а потому никакие падения и переполнения памяти ей не страшны. Она может быть как пустой, так и занимать сотни гигабайт в определенные моменты времени. Ее дело получить сообщение и предоставить на чтение одному из обработчиков в порядке получения, когда он будет готов. Помимо прочего она требует, чтобы обработчик сообщил об успешной обработке -- если этого не произошло, то сообщение через некоторое время снова появится в очереди (и этот процесс повторится конфигурируемое число раз). В Azure есть два решения на этот счет Service Bus Queue и Storage Queue. Все они отвечают этому паттерну, дают гарантии, хоть и реализуют их различными способами. По описанию задачи вам хватит Storage Queue.

Дело же обработков -- взять сообщение, обработать его и сообщить очереди, что про данное сообщение можно забыть. Исторически в Azure на этот счет было одно решение Azure WebJobs. Разработанный MS SDK очень удобен. Все что вам нужно написать -- это функция вроде этой в WebJob проекте в Visual Studio:

public static void ProcessQueueMessage([QueueTrigger("myqueue")] string logMessage)
{
    // логика по обработке данных
}

Здесь ProcessQueueMessage -- произвольное название функции, QueueTrigger -- триггер, говорящий WebJob, что необходимо вызывать эту функцию при появлении сообщения в очереди, myqueue -- название очереди, string -- тип сообщения (может быть пользовательский класс, в который будет производиться попытка десериализовать сообщение), само указание на Storage находится в конфигурационном файле. Прочтите эту статью, все безумно просто. При каждом появлении сообщении в этой очереди эта фунуция будет вызываться автоматически, сообщение будет десериализовываться в указанный тип и начинаться выполнение. Если сообщение завершиться исключением, то обработка будет считаться неуспешной, иначе -- сообщение будет окончательно извлечено из очереди. Параллельно могут исполнаться до 8 обработчиков на одной машине, число машин можно шкалировать с помощью автоматических правил. Вот так просто. Я заметил, что вы писали код на Node.js, тогда вам точно стоит посмотреть в сторону Functions -- эта среда исполнения там поддерживается. Подход к написанию кода несколько другой, но по сути, если я правильно понимаю, Functions в итоге создают WebJob, инъектирует в него код своего движка, который уже внутри вызывает ваш.

Надеюсь стало понятнее. На тему запроса, а вы пробовали запускать отсылаемый запрос вручную через SQL Server Management Studio или Visual Studio? Сколько он выполняется? Может вся проблема в производительности базы или индексах? Но кажется это уже офтопик к основной теме обсуждения и ее стоит вынести наружу.

Answer 13

Большое спасибо за столь подробный ответ. Направление куда "копать" понял. Буду читать про WebJob + Storage Queue.

Несколько вопросов:

0. Вы говорите, можно через сервисы "отловить" инсерты со Stream Analytic - у меня не получилось. В INSERT скрипте ничего не происходит в момент добавления записи. На своём корявом английском задал вопрос на англоязычном форуме - но пока непонятно Может вы знаете, как их отловить?.

1. Как всё же вы бы обновляли данные в базе? Потому что помимо действий нужно их ещё записать в таблицу. 

2. Что происходит со Storage Queue, если не извлекать сообщения из очереди? Разрастется до какого-то предела (это оффтоп, конечно)? 

3. Можно ли с WebJob "достучаться" до внутренних объектов Azure MS или Azure SQL? Я имею в виду, можно ли работать с базой как я работаю в шедулере или обработчике? table1 = require("table1") (и не прописываю строку подключения). 

Answer 14

0. Попробуйте подключиться к базе с помощью SQL Server Profiler и посмотреть, какие запросы идут в ее сторону. Скорее всего он высылает их стандартным способом.

1. Как я уже писал, вам стоит написать решение на базе Entity Framework или ADO.NET в Web Job собственный код (как будто это совершенно внешнее приложение, никак несвязанное с Azure, со строками соединения и т.п.). Про Node.JS ничего не знаю, подсказать не могу. Вам надо просто отослать запрос MERGE (или последовательно SELECT, INSERT или DELETE) с данными в сторону базы любыми известными вами способами.

2. 500 TB, если я правильно понимаю документацию.

3. Тут работа полностью как с внешним ресурсом средствами выбранного SDK. Строка подключения в любом случае где-нибудь задана, хотя может и неявно.