Во второй части статьи рассказывалось о механизмах обнаружения ошибок в процессе обработки.
Обработка завершилась с ошибкой, что делать дальше? Вполне возможно, что потеряна связь с одним из узлов кластера или временно недоступна база данных. В этом случае, нельзя с уверенностью сказать, какие операции выполнились успешно, а какие — нет. Если все операции в цепочке повторно применимы (идемпотентны), например установка флага, то можно просто перезапустить обработку. Если нет, то на помощь приходят механизмы транзакций 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, транзакции он не поддерживает.
Автор: xlix123
