IBM InfoSphere Streams предоставляет собой мощное и масштабируемое решение для обработки непрерывного потока миллионов кортежей данных в секунду. Этот механизм потоков обеспечивает надежную и высокомасштабируемую среду выполнения, которая обрабатывает аналитическую логику на сотнях узлов и процессорных ядер. Это предоставляет пользователю возможность сосредоточиться на применении данных, не думая о том, как и где это делается.
В своей типичной области применения InfoSphere Streams помогает динамически обрабатывать входящие данные и за считанные секунды реагировать на события, вместо того чтобы делать это постфактум и с задержкой в несколько минут или даже часов. На рисунке 1 показана архитектура системы, в которой Streams помогает обработать данные по 6 млрд. вызовов в день в сети мобильный связи.Рисунок 1. Streams помогает выявить проблемы в сотовой сети
Время обработки и загрузки этих записей в систему BI и системы учета сократилось с 12 часов до нескольких секунд, и теперь можно наблюдать активность клиентов и качество работы сети в режиме реального времени.
Система Streams предназначена для динамической обработки данных, но ее можно использоваться и в целях моделирования. В следующей главе описывается, как настроить поток данных таким образом, чтобы после обработки они вновь возвращались в обрабатывающий механизм. Для этого нужно решить несколько проблем, которые на первый взгляд не очевидны; ниже мы поговорим о них подробнее.
Основы Streams
Поток ― это непрерывная последовательность данных, направляющихся от источника к месту назначения. Данные перемещаются в форме кортежей, которые состоят из фиксированного набора атрибутов. Поток можно сравнить с содержимым базы данных. Там данные хранятся в форме строк и столбцов. Столбцы ― это индивидуально определенные элементы, которые составляют запись данных (или строку). В потоках такие записи называются кортежами, а столбцы ― атрибутами.
Чтобы понять идею обработки потоков, важно усвоить четыре термина: адаптеры источников/приемников, операторы, язык программирования потоков (Streams Programming Language - SPL) и среда выполнения потоков.
Рисунок 2. Среда Streams
Источник потока называется источником, а место назначения ― приемником. Такие приемники или источники могут быть файлами, базами данных, сокетами TCP/IP или любыми другими специальными программами на языке Java™ или C++, которые создают или сохраняют кортежи.
Операторы позволяют обрабатывать поток кортежей с целью фильтрации, суммирования, слияния, преобразования или выполнения других, более сложных математических функций, таких как быстрое преобразование Фурье. Они принимают в качестве входа один или более потоков, выполняют обработку кортежей и атрибутов и выдают на выходе один или несколько потоков.
Streams-приложение представляет собой совокупность таких операторов, соединенных потоками. Такие приложения пишутся на декларативном языке программирования типа SQL. В отличие от более универсальных языков программирования, таких как C, C++ или Java, декларативный язык программирования описывает не то, что должно быть сделано, а то, как должны выглядеть ожидаемые результаты. Компилятор преобразует описание результатов в программный код, выполняющий операции, необходимые для их получения.
Во время выполнения приложения для обработки потоков среда выполнения заботится о надежном выполнении адаптеров приемника и источника, операторов и об их распределении между доступными процессорными ядрами и узлами для достижения оптимальной пропускной способности.
Streams-приложения можно использовать для решения задач анализа потоков по мере прохождения кортежей. Имеющаяся информация используется в момент ее получения, так что пользователь видит ее почти в режиме реального времени, не дожидаясь следующего окна, как при традиционной обработке.
Моделирование с помощью Streams
В ходе недавнего исследования была разработана модель для имитации изменений условий дорожного движения по сети улиц (ремонтные работы, аварии, введение дополнительных полос, ограничение скорости и т.п.) и определения их влияния на пропускную способность и длину пробок. Среда Streams была выбрана за высокую масштабируемость при параллельной обработке, что идеально подходит для модели дорожного движения Нагеля-Шрекенберга.
Эта модель имитирует движение каждого автомобиля, грузовика или мотоцикла. Один цикл моделирования соответствует 1 секунде реального времени. Дорога делится на секторы длиной по 7,5 м, и каждое транспортное средство перемещается по правилам дорожного движения и ограничения скорости с возможностью ускорения или замедления.
Для простоты и управляемости вся сеть улиц не моделируется как один большой объект. Вместо этого каждый участок дороги между перекрестками, въездами на автомагистрали и съездами с них, или пешеходными переходами представлен отдельной моделью Нагеля-Шрекенберга, а затем каждый такой участок увязывается с другими. Транспортные средства могут переходить с одного участка на другой в конце каждого цикла моделирования.
Рисунок 3. Связанные участки модели Нагеля-Шрекенберга
Почему циклы должны быть синхронизированы
Каждая улица моделируется одним Streams-приложением. Среда Streams распределяет вычислительные задачи между имеющимися ресурсами компьютеров и процессоров. Важно, чтобы каждое приложение завершало цикл до начала нового, потому что приложения, обрабатывающие короткие улицы, выполняют больше циклов, и без синхронизации с коротких улиц будет выезжать гораздо больше машин, чем с других. Этой проблемы нет, если за каждой улицей следует только одна новая. Но как только появляются въезды/създы и перекрестки, как в реальной жизни, разные скорости циклов начинают создавать проблемы.
Въезд на участки дорог и выезд с них
Каждое Streams-приложение на входах и выходах потоков кортежей может использовать специальные адаптеры приемников и источников. В нашем случае адаптер источника должен дождаться прибытия транспортных средств с предыдущего участка дороги, а адаптер приемника ― управлять транспортными средствами, покидающими участок при переезде на следующий. Стандартные операторы источника и приемника могут считывать или записывать данные в различных форматах в файлы, базы данных или сокеты TCP/IP и из них, но не имеют необходимых средств для синхронизации циклов моделирования всех потоковых приложений. Мы создали отдельный сервер для синхронизации своих адаптеров приемников и источников: адаптеры приемников и источников TCP/IP.
Реализация синхронизации
В начале каждого цикла все автомобили вводятся в Streams-приложение адаптером источника, прослушивающим сокет TCP/IP. Для синхронизации частоты циклов для всех приложений Streams каждое приложение должно останавливать обработку входных данных до тех пор, пока все приложения не закончат расчет предыдущего цикла. Для этого можно использовать оператор сортировки Streams. Он сортирует кортежи в так называемом окне и выпускает все отсортированные кортежи сразу. В нашем случае адаптер источника выпускает на участок дороги специальный автомобиль, называемый пунктором, который указывает, что все транспортные средства приняты. Оператор punctor дожидается такого автомобиля-пунктора и генерирует маркер окна, который сигнализирует оператору сортировки, что все транспортные средства данного цикла моделирования прибыли, и их можно выпускать. Затем, как показано на следующем рисунке, каждый автомобиль перемещается по правилам модели Нагеля-Шренкенберга.
Рисунок 4. Связь между Stream-приложениями и Java-потоками
В конце обработки поток делится на две части. Все выезжающие транспортные средства, включая сам автомобиль-пунктор, направляются через TCP-приемник в синхронизатор; а транспортные средства, которые остаются, непосредственно передаются оператору punctor.
Синхронизатор и Streams-приложения взаимодействуют через TCP-сокеты. Streams-приложения используют оператор приемника TCP/IP для записи данных об автомобилях в сокет в виде строк со значениями разделенными запятыми (CSV). Синхронизатор реализован как отдельный сервер TCP/IP, написанный на языке Java, который получает данные об автомобилях из сокета. Он отслеживает все запущенные Streams-приложения и знает, какие из них уже выпустили свои автомобили-пункторы. Когда все Stream-приложения закончили свои циклы моделирования, подается сигнал генератору транспортных средств, который отправляет все автомобили, оставшиеся на участке дороги, в адаптер источника следующего участка.
Заключение
InfoSphere Streams не ограничивается только обработкой в режиме реального времени кортежей данных, а может использоваться и в качестве инфраструктуры параллельной обработки данных в целях моделирования. Его среда выполнения обеспечивает высокую готовность, планирование и параллельную обработку, предоставляя пользователю возможность сосредоточиться на применении своих данных. Среда выполнения Streams содержит набор готовых операторов и функций, которые позволяют лучше выполнить все требования, предъявляемые к обработке. В тех случаях, когда этого недостаточно, к ним можно легко добавить операторы и функции собственной разработки.