Изучаем Storm Framework. Часть I

Storm это система ориентированная на распределенную обработку больших потоков данных, аналогичная Apache Hadoop, но в реальном времени.

Ключевые особенности Storm:

• Масштабируемость. Задачи обработки распределяются по узлам кластера и потокам на каждом узле.
• Гарантированная защита от потери данных.
• Простота развертывания и спровождения.
• Восстановление после сбоев. Если какой либо из обработчиков отказывает, задачи переадресуются на другие обработчики.
• Возможность написания компонентов не только на Java. Простой Multilang protocol с использованием JSON объектов. Есть готовые адаптеры для языков Python, Ruby и Fancy.

В первой части рассматриваются базовые понятия и основы создания приложения c использованием Storm версии 0.8.2.

Элементы Storm

Tuple
Элемент представления данных. По умолчанию может содержать Long, Integer, Short, Byte, String, Double, Float, Boolean и byte[] поля. Пользовательские типы используемые в Tuple должны быть сериализуемыми.

Stream
Последовательность из Tuple. Содержит схему именования полей в Tuple.

Spout
Поставщик данных для Stream. Получает данные из внешних источников, формирует из них Tuple и отправляет в Stream. Может отправлять Tuple в несколько разных Stream. Есть готовые для популярных систем обмена сообщениями: RabbitMQ / AMQP, Kestrel, JMS, Kafka.

Bolt
Обработчик данных. На вход поступают Tuple. На выход отправляет 0 или более Tuple.

Topology
Совокупность элементов с описанием их взаимосвязи. Аналог MapReduce job в Hadoop. В отличии от MapReduce job — не останавливается после исчерпания входного потока данных. Осуществляет транспорт Tuple между элементами Spout и Bolt. Может запускаться локально или загружаться в Storm кластер.

Пример использования

Задача

Есть поток данных о телефонных вызовах Cdr. На основании source номера определяется id клиента. На основании destination номера и id клиента определяется тариф и считается стоимость звонка. Каждый из этапов должен работать в несколько потоков.
Пример будет запускаться на локальной машине.

Реализация

Для начала просто распечатаем входные данные BasicApp.

Создаем новую Topology:

TopologyBuilder builder = new TopologyBuilder(); 

Добавляем Spout CdrSpout генерирующий входные данные:

builder.setSpout("CdrReader", new CdrSpout()); 

Добавляем Bolt с двумя потоками и указываем что на вход подается выходной поток CdrReader. shuffleGrouping означает что данные из CdrReader подаются на случайно выбранный Bolt.

builder.setBolt("PrintOutBolt", new PrintOutBolt(), 2).shuffleGrouping("CdrReader"); 

Конфигурируем и запускам локальный Storm кластер:

Config config = new Config(); // Конфигурация кластера по умолчанию config.setDebug(false);  LocalCluster cluster = new LocalCluster(); // Создаем локальный Storm кластер cluster.submitTopology("T1", config, builder.createTopology()); // Стартуем Topology Thread.sleep(1000*10);  cluster.shutdown(); // Останавливаем кластер 

На выходе получаем примерно следующее:

 OUT>> [80]Cdr{callSource='78119990005', callDestination='8313610698077174239',  callTime=7631, clientId=0, price=0} OUT>> [78]Cdr{callSource='78119990006', callDestination='2238707710336895468',  callTime=20738, clientId=0, price=0} OUT>> [78]Cdr{callSource='78119990007', callDestination='579372726495390920',  callTime=31544, clientId=0, price=0} OUT>> [80]Cdr{callSource='78119990006', callDestination='2010724447342634423',  callTime=10268, clientId=0, price=0} 

Число в квадратных скобках — Thread Id, видно что обработка ведется параллельно.

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

Для этого в Storm предусмотрен интерфейс CustomStreamGrouping.
Добавим в проект CdrGrouper. Его задача — отправлять Tuple с одинаковыми source номерами на один и тот же Bolt. Для этого в CustomStreamGrouping предусмотрено два вызова:
prepare — вызывается перед первым использованием:

@Override     public void prepare(WorkerTopologyContext workerTopologyContext, GlobalStreamId globalStreamId, List<Integer> integers) {         tasks = new ArrayList<>(integers); // Запоминаем номера Bolts     } 

и chooseTasks — где на вход подается список из Tuple, а возвращается список состоящий из номеров Bolt’ов для каждой позиции в списке Tuple:

@Override     public List<Integer> chooseTasks(int i, List<Object> objects) {         List<Integer> rvalue = new ArrayList<>(objects.size());         for(Object o: objects) {             Cdr cdr = (Cdr) o;              rvalue.add(tasks.get(Math.abs(cdr.getCallSource().hashCode()) %                              tasks.size()));         }         return rvalue;     } 

Заменим shuffleGrouping на CdrGrouper BasicGroupApp:

builder.setBolt("PrintOutBolt", new PrintOutBolt(), 2).                 customGrouping("CdrReader", new CdrGrouper()); 

Запустим и убедимся что работает как задумано:

 OUT>> [80]Cdr{callSource='78119990007', callDestination='3314931472251135073',  callTime=17632, clientId=0, price=0} OUT>> [80]Cdr{callSource='78119990007', callDestination='4182885669941386786',  callTime=31533, clientId=0, price=0} 

Далее в проект добавляем:
ClientIdBolt — определяет id клиента по source номеру.
ClientIdGrouper — Группирует по id клиента.
RaterBolt — занимается тарификацией.
CalcApp — окончательный вариант программы.

Если тема будет интересна, то в следующей части надеюсь рассказать о механизмах защиты от потери данных и запуске на реальном кластере. Код доступен на GitHub.

PS. Из песни конечно слова не выкинешь, но название обработчика данных «Bolt» несколько смущает 🙂