Обзор технологии Microsoft StreamInsight. Часть 2

Адаптеры

Как было замечено ранее, адаптеры служат для преобразования потоков данных. Адаптеры делятся на входные и выходные. Входные адаптеры преобразуют данные полученные с источников, в формат понятный StreamInsight, а выходные адаптеры преоразуют данные в формат понятный конечным получателям.

Основные задачи, которые нужно решить при разработке адаптеров:

Определить тип адаптера.

Входной или выходной.

Определить тип событий.

Адаптер может быть типизированным либо нетипизированным. Структура событий для типизированных адаптеров известна заранее (описана в виде класса), а для нетипизиванных неизвестна (например, данные получаются с использованием DataReader из Microsoft SQL Server).

Определить модель событий.

Point, Interval или Edge. Рекомендуется создавать отдельный адаптер для каждой модели событий.

Выбрать подходящий базовый класс для адаптера.

Существуется несколько базовых классов адаптеров для каждого типа адаптера и модели события.

Тип адаптера	Базовый класс входного адаптера	Базовый класс выходного адаптера
Типизированный, Point	TypedPointInputAdapter	TypedPointOutputAdapter
Нетипизированный, Point	PointInputAdapter	PointOutputAdapter
Типизированный, Interval	TypedIntervalInputAdapter	TypedIntervalOutputAdapter
Нетипизированный, Interval	IntervalInputAdapter	IntervalOutputAdapter
Типизированный, Edge	TypedEdgeInputAdapter	TypedEdgeOutputAdapter
Нетипизированный, Edge	EdgeInputAdapter	EdgeOutputAdapter

Создать фабрики для входных и выходных адаптеров

Тип адаптера	Базовый класс фабрики входного адаптера	Базовый класс фабрики выходного адаптера
Типизированный	ITypedInputAdapterFactory	ITypedOutputAdapterFactory
Нетипизированный	IInputAdapterFactory	IOutputAdapterFactory

Далее перечислены основные обязанности фабрики:

• Разделяет ресурсы между сходными адаптерами, различающимися только моделью событий.
• Предоставляет интерфейс Create() и Dispose(). С их помощью адаптеры работают с событиями.
• Автоматически создает CTI события с учетом пользовательских настроек.

Разработчик адаптера должен имплементировать методы Start() и Resume() базового класса, в которых производится преобразование событий.

Базовые классы предоставляют методы Enqueue() и Dequeue(), которые должны использоваться во входном и выходном адаптерах соответственно для обработки событий. Например, входной адаптер преобразует событие в формат понятный StreamInsight и помещает его в очередь, используя метод Enqueue().

Очень подробно разработка адаптеров описана в разделе Creating Input and Output Adapters документации StreamInsight.

Запросы

Пожалуй, наиболее интересным с точки зрения разработчика элементом StreamInsight являются запросы.

Запросы StreamInsight представляют собой не что иное, как LINQ запросы к потокам данных.

Простейший запрос мог бы выглядеть следующим образом:

var queryOutput = from e in input

select e;

Такой запрос просто возвращает все полученные события.

StreamInsight запросы предоставляют следующие возможности:

• Проекции (Project)
• Фильтры (Filter)
• Корелляция потоков (Join)
• Объединения (Union)
• Агрегация (Aggregation)
• Оконные операции

Рассмотрим каждую из возможностей:

Проекции

Проекции используется в случаях, когда нужно:

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

Следующий запрос использует пользовательскую функцию для получения нового поля:

var queryOutput = from e in input

select
new { e.Lane, e.TagId,

VehicleType = TollPointEvent.VehicleTypeName(e.VehicleTypeId) };

Фильтры

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

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

var queryOutput = from e in input

where e.VehicleTypeId == 2

select
new { e.Lane, e.TagId, VehicleType = TollPointEvent.VehicleTypeName(e.VehicleTypeId) };

Корелляция потоков

Используется операция Join, хорошо знакомая разработчикам, использующим SQL. Разница в том, что объединятся не все события, а только те, которые произошли в один и тот же промежуток времени.

Следующий запрос выполняет cross join двух потоков событий.

var queryOutput = from nbv in northboundVehicles

from sbv in lane0Vehicles

select
new

{

ExitNorth = nbv.ExitGate,

NorthVehicle = nbv.TagId,

TollPointId = sbv.TollPointId

};

Объединение

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

Следующий запрос объединяет события их двух разных потоков.

var northboundVehicles = from e in input

where e.DirectionId == 0

select e;

var lane0Vehicles = from e in input

where e.Lane == 0

select e;

var queryOutput = northboundVehicles.Union(lane0Vehicles);

Безусловно, можно объединять более двух потоков, последовательно вызывая Union()

Агрегация

Агрегация позволяет вычислять некоторую функцию по набору событий. К таким фукнциям относятся: Avg, Sum, Count и т.д. Также можно использовать пользовательские функции. Функции агрегации можно применять только к наборам событий, в данном случае — только к окнам.

Окна позволяют объединять события (для последующей обработки) на интервалах, которые могут задаваться временными или количественными характеристиками.

Для окон поддерживаются следующие виды операций:

• Агрегация.
• TopK — ранжирование.
• Пользовательские операторы.

В StreamInsight существует четыре типа окон:

• Временные окна: Hopping Window и Tumbling Window
• Окна моментальных снимков: Snapshot Window
• Окна количества: CountByStartTime Window

Hopping Window (прыгающее окно)

Для прыгающего окна определяются два параметра: H – размер «прыжка» и размер окна S. Новое окно создается через каждые H моментов времени (окно «прыгает»), а размер этого окна равен S.

Если H равно S, такое окно называется Tumbling Window (окно вращения).

Следующий запрос вычисляет сумму по всем событиям, произошедшим за последний час, каждые 10 минут.

var hoppingAgg = from w in inputStream.HoppingWindow(TimeSpan.FromHours(1),

TimeSpan.FromMinutes(10),

HoppingWindowOutputPolicy.ClipToWindowEnd)

select new { sum = w.Sum(e => e.i) };

Обратите внимание, что при создании окна используется параметр, задающий политику HoppingWindowOutputPolicy.ClipToWindowEnd. Интересным является тот факт, что в текущей версии StreamInsight существует всего один вариант задания данной политики (правда он называется по-разному для каждого типа окна).

Snapshot Window (Окно моментального снимка)

Параметры окон моментального снимка определяются только событиями в потоке. Появление нового события или окончание старого приводит к окончанию текущего окна и созданию нового (если какие-либо события еще не закончены). Рисунок хорошо иллюстрирует сказанное:

Следующий запрос создает окно мементального снимка:

var snapshotAgg = from w in inputStream.SnapshotWindow(WindowInputPolicy.ClipToWindow,

SnapshotWindowOutputPolicy.Clip)

select new { sum = w.Sum(e => e.i) };

CountByStartTime Window (Окно количества)

Размер окон количества зависит от количества событий с разными временными отметками (точнее, с разным временем начала события). Если все события имели разное начальное время, то окно будет содержать ровно N (параметр окна количества) событий.

Окно количество можно задать следующим запросом:

var agg = from w in inputStream.CountByStartTimeWindow(10, CountWindowOutputPolicy.PointAlignToWindowEnd)

select new { sum = w.Sum(e => e.i) };

Для более подробного ознакомления с окнами событий, я рекомендую обратиться к соответствующему разделу документации StreamInsight (Developer’s Guide->Writing Query Templates in LINQ->Using Event Windows).

Этапы разработки

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

1. Определить параметры событий и создать класс нагрузки
2. Создать входные адаптеры и фабрики входных адаптеров для каждого типа источников данных
3. Создать шаблоны StreamInsight запросов с использованием LINQ
4. Создать выходные адаптеры и фабрики выходных адаптеров для каждого типа получателей событий
5. Связать все компоненты системы вместе

Также нужно отметить, что движок StreamInsight может работать в нескольких режимах:

• Как отдельный сервис
• Как встроенная часть приложения

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

Заключение

Технология StreamInsight, как минимум, заслуживает внимания, поскольку она предоставляет новый способ анализа данных. Причем, скорость обратки событий и парадигма в целом не могут не радовать.

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

• Непонятная ценовая политика. Во время подготовки статьи я не смог найти бесплатной версии StreamInsight без ограничения по времени или возможности купить StreamInsight отдельно от SQL Server 2008 R2. (Хотя, возможно, я плохо искал)
• Не понятно, каким образом аналитик должен писать LINQ запросы. Как пока известно, планируется разработка специального языка StreamInsight запросов, понятного аналитику, а не только разработчику.

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

Материалы для дальнейшего изучения

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

Techdays.ru – материалы на русском

(http://www.techdays.ru/Search.aspx?Tag=SQL%2bServer%2b2008%2bR2)

SQL Server 2008 R2 Update for Developers Training Kit – крайне полезные обучающие материалы
(http://www.microsoft.com/downloads/details.aspx?familyid=FFFAAD6A-0153-4D41-B289-A3ED1D637C0D&displaylang=en)

Документация по StreamInsight (поставляется вместе с StreamInsight)