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

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

Когда говорят о характеристиках транзакций, тут же всплывает термин ACID:

• Atomicity (атомарность). Все изменения произведенные в системе на протяжении транзакции, либо применяются полностью, либо не применяются совсем.
• Consistency (cогласованность). Транзакция переводит систему из одного непртиворечивого состояния в другое.
• Isolation (изолированность). Параллельно выполняемые транзакции не оказывают влияние на результат работы друг друга.
• Durability (надежность). Зафиксированные транзакцией изменения гарантированно остаются в системе.

Consistency и Durability в большей степерни относятся к базам данных. Нас будут интересовать Atomicity и Isolation.

В версии 0.8.0 в Storm появилась подсистема Trident — аналог Apache Pig. В нее же перекочевал функционал Transactional topology.

Транзакции в Storm

Atomicity

В Topology создается объект реализующий интерфейс State, инкапсулирующий работу с БД. Входные данные, поступающие в Spout, разбиваются на Tuple и собираются в пакеты (batch). Batch ассоцируется с уникальным transaction id. Tuple образующие batch могут обрабатываться параллельно.
В конце цепочки обработки, набор Tuple, относящихся к одной транзакции, передается в метод updateState класса, реализующего интерфейс StateUpdater, который и призводит модификацию State. В случае успешного завершения, Spout получает уведомление об успехе обработки batch’a. В случае ошибки, Spout должен передать на обработку весь batch повторно.
Таким образом Storm гарантирует, что Batch будет зафиксирован в БД полностью и только один раз.

Isolation

Storm гарантирует, что Batch’и передаются в StateUpdater строго последовательно, в порядке возрастания transaction id. То есть Batch #2 будет зафиксирован только после успешной фиксации Batch’а #1.

Реализация

Spout с поддержкой транзакций должен реализовывать интерфейс ICommitterTridentSpout<TransactionMetadata>. TransactionMetadata — любой класс, содержит данные для генерации Batch’ей и генерации следующей транзакции: TxMeta.

public class TxMeta {     private int start;     private int count;      public TxMeta(int start, int count) {         this.start = start;         this.count = count;     } // Skipped getters  } 

Класс реализующий интерфейс ITridentSpout.BatchCoordinator<TransactionMetadata> инициализирует TransactionMetadata при создании транзакции и отвечает на запрос готовы ли данные для следующей транзакции: TridentTxSpout. Создается в единственном экземпляре для каждой Topology.

    static class BCoordinator implements BatchCoordinator<TxMeta> {         private static final int TRANSACTION_COUNT = 5;         private static final int TRANSACTION_ELEMENT_COUNT = 5;  //TxMeta - метаданные предыдущей транзакции         @Override         public TxMeta initializeTransaction(long l, TxMeta txMeta) {              if(txMeta != null) {              System.out.println(String.format("Initializing transaction id: %08d, "             + "start: %04d, count: %04d", l, txMeta.getStart() +                 txMeta.getCount(), txMeta.getCount()));              return new TxMeta(txMeta.getStart() + txMeta.getCount(),                     TRANSACTION_ELEMENT_COUNT);             } else {                 return new TxMeta(0, TRANSACTION_ELEMENT_COUNT);             }         }  // Готовы ли данные для следующей транзакции         @Override         public boolean isReady(long l) {             if(l <= TRANSACTION_COUNT) {                 System.out.println("ISREADY " + l);                 return true;             }             return false;         }     } 

Класс реализующий интерфейс ICommitterTridentSpout.Emitter формирует Batch. В случае ошибки в обработке Batch’a, формирует Batch повторно.
Важно — повторно сформированный Batch должен содержать точно такой же набор Tuple, что и оригинальный.

static class BEmitter implements Emitter { // Формирует Batch по информации из TransactionMetadata         @Override         public void emitBatch(TransactionAttempt transactionAttempt,                                Object coordinatorMeta,                               TridentCollector tridentCollector) {              TxMeta txMeta = (TxMeta) coordinatorMeta;              System.out.println("Emitting transaction id: " +                     transactionAttempt.getTransactionId() + " attempt:" +                     transactionAttempt.getAttemptId()             );             for(int i = 0; i < txMeta.getCount(); ++i) {                 tridentCollector.emit(new Values("TRANS [" +                         transactionAttempt.getAttemptId() +                          "] [" + (txMeta.getStart() + i) + "]")                 );             }         }  // Транзакция  успешно закоммичена в State        @Override         public void success(TransactionAttempt transactionAttempt) {             System.out.println("BEmitter:Transaction success id:" +                                           transactionAttempt.getTransactionId());         }  // Попытка коммита транзакции в State         @Override         public void commit(TransactionAttempt transactionAttempt) {             System.out.println("BEmitter:Transaction commit id:" +                                          transactionAttempt.getTransactionId());         }     } 

Класс реализующий интерфейс State в нашем случае драйвер БД: TxDatabase.

public class TxDatabase implements State { // Вызывается при начале транзакции в БД     @Override     public void beginCommit(Long txId) {         System.out.println("beginCommit [" + Thread.currentThread().getId() + "] " + txId);     }  // Вызывается для коммита транзакции в БД     @Override     public void commit(Long txId) {         System.out.println("commit [" + Thread.currentThread().getId() + "] " + txId);     } } 

Класс наследующий BaseStateUpdater<S extends State>, вносит изменения в State (БД): TxDatabaseUpdater

public class TxDatabaseUpdater extends BaseStateUpdater<TxDatabase> {     int count;      // Вносит изменения в БД     @Override     public void updateState(TxDatabase txDatabase,                              List<TridentTuple> tridentTuples,                             TridentCollector tridentCollector) {          // Эмуляция сбоя транзакции         if(++count == 2) throw new FailedException("YYYY");           for(TridentTuple t: tridentTuples) {             System.out.println("Updating: " + t.getString(0));         }     } } 

Класс реализующий интерфейс StateFactory, создает экземпляры State: TxDatabaseFactory.

Собираем все вместе TridentTransactionApp:

public class TridentTransactionApp {     public static void main( String[] args ) throws Throwable     {         Logger.getRootLogger().setLevel(Level.ERROR);  // Создаем топологию         TridentTopology tridentTopology = new TridentTopology(); // Добавляем наш Spout         tridentTopology.newStream("TridentTxSpout", new TridentTxSpout()). // Обработка Tuple пойдет параллельно - OpPrintout просто печатает записи                 shuffle().each(new Fields("msg"), new OpPrintout()).                 parallelismHint(2). // Сливаем результаты параллельной обработки в один поток                 global(). // Записываем изменения в State (БД)                 partitionPersist(new TxDatabaseFactory(),                         new Fields("msg"), new TxDatabaseUpdater()); // Skipped         LocalCluster cluster = new LocalCluster();         cluster.submitTopology("T2", config, tridentTopology.build());         Thread.sleep(1000*100);         cluster.shutdown();     } } 

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

PS. Раскрыть все возможности Storm в рамках статей невозможно, материала хватит на целую книгу. Надеюсь мне удалось показать ключевые возможности фреймворка и возможности его применения в реальных проектах.
По поводу развертывания кластера — недавно наткнулся на отличную статью. Не вижу смысла повторяться. Развернуть Storm в production действительно несложно.

PPS. В Hadoop существует аналог on-line обработки Storm — Hadoop Streaming, но в отличии от Storm, транзакции он не поддерживает.