Новости

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

Основные стриминговые концепции

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


Если забыть об исполнителях и рассматривать только объекты, определяемые пользователем, мы получим новую диаграмму справа — более чистое (более абстрактное) краткое представление стриминговой системы. Эта диаграмма (назовем ее логическим планом) является высокоуровневой абстракцией, на которой изображены компоненты и структуры системы, а также логические потоки данных в них. Из этой диаграммы можно видеть, как объект-источник и объект-оператор соединяются через поток для формирования стримингового задания. Следует понимать, что поток — не что иное, как непрерывная передача данных от одного компонента к другому.

Подробнее о концепциях

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


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

Последовательность выполнения стримингового задания

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

• Блок чтения данных получает данные с датчика и сохраняет события в очереди. Это источник.
• Счетчик автомобилей отвечает за подсчет автомобилей в потоке. Это оператор.
• Непрерывное перемещение данных от источника к оператору — поток событий автомобилей.

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

В этой главе такая система будет описана более подробно. Она будет выполняться на ваших локальных компьютерах с двумя терминалами: один получает ввод пользователя (левый столбец), а другой — вывод задания (правый столбец).

Первое стриминговое задание

Стриминговое задание создается средствами Streamwork API и включает следующие шаги.
1. Создание класса задания.
2. Построение источника.
3. Построение оператора.
4. Соединение компонентов.

Первое стриминговое задание: создание класса события
Событие — один фрагмент данных в потоке, который должен обрабатываться заданием. Во фреймворке Streamwork класс API Event отвечает за хранение или инкапсуляцию пользовательских данных. Аналогичные концепции существуют и в других стриминговых системах.

В задании каждое событие представляет один тип автомобилей. Для простоты будем считать, что каждый тип представляет собой строку (например, car или truck). В нашем примере будет использоваться класс события VehicleEvent, производный от класса Event из API. Каждый объект VehicleEvent содержит информацию об автомобиле, которую можно получить вызовом функции getData().


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


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

Что такое перехватчик жизненного цикла?
Перехватчиками жизненного цикла называются методы, которые вызываются по определенному повторяемому шаблону фреймворком, которому они принадлежат. Как правило, эти методы позволяют разработчикам настроить поведение приложения в фазах жизненного цикла фреймворка, на основе которого строится приложение. В случае фреймворка Streamwork имеется перехватчик жизненного цикла (или метод), называемый getEvents(). Он многократно вызывается фреймворком, чтобы вы могли получить внешние данные. Перехватчики жизненного цикла позволяют разработчикам писать логику, которая для них важна, и поручить рутинную работу фреймворку.

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


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


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

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

В компоненте VehicleCounter карта <vehicle, count> используется для хранения счетчиков типов автомобилей в памяти. Она обновляется соответствующим образом при получении нового события. В стриминговом задании счетчик автомобилей представляет собой оператор, который подсчитывает события. Этот оператор завершает задание и не генерирует вывод для последующих операторов.

Первое стриминговое задание: сборка задания
Чтобы собрать стриминговое задание, необходимо добавить источник SensorReader и оператор VehicleCounter и соединить их. В классах Job и Stream, которые мы построили для вас, присутствуют перехватчики:

• Job.addSource() добавляет источник данных в задание.
• Stream.applyOperator() добавляет оператор в поток.

Следующий код выполняет описанные выше шаги:

Выполнение задания

Все, что вам понадобится для выполнения задания, — машина с операционной системой Mac, Linux или Windows с доступом к терминалу (командной строке в Windows). Вам также понадобятся инструменты для компиляции и выполнения кода: git, пакет Java Development Kit (JDK) 11, Apache Maven, Netcat (или Nmap в Windows). После установки всех инструментов вы сможете загрузить код и откомпилировать его:

$ git clone https://github.com/nwangtw/GrokkingStreamingSystems.git
$ cd GrokkingStreamingSystems
$ mvn package

Команда mvn генерирует файл target/gss.jar. Наконец, для запуска стримингового задания вам понадобятся два терминала: для запуска задания и для отправки данных, обрабатываемых заданием.


Откройте новый терминал (терминал ввода) и выполните следующую команду (обратите внимание: в Mac и Linux используется команда nc, а в Windows — команда ncat):

$ nc -lk 9990

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

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

$ java -cp target/gss.jar com.streamwork.ch02.job.VehicleCountJob

Ход выполнения задания

После запуска задания введите в терминале ввода строку car и нажмите Ввод. Счетчик выводится в терминале задания.


Если теперь ввести в терминале ввода строку truck, в терминале задания будут выведены счетчики для car и truck.


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

Внутри ядра

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


Задания и компоненты не выполняются сами по себе. Ими управляет стриминговое ядро. Заглянем под капот и посмотрим, как задание выполняется в ядре Streamwork. Всего существуют три активные части (на этом этапе), и мы рассмотрим их поочередно: исполнитель-источник, исполнитель-оператор и блок запуска задания.


Внутри ядра: исполнители-источники
Во фреймворке Streamwork, который мы построили для вас, исполнитель-источник непрерывно управляет получением данных от источников. Для этого он выполняет бесконечные циклы, которые извлекают внешние данные для включения в исходящую очередь внутри стримингового задания. Несмотря на то что для блока Выход предусмотрен вариант Да, он никогда не будет реализован.


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


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