Apache Spark Streaming, Как обрабатывать отказы зависимостей по нисходящей линии

Я пытаюсь понять, как сделать приложение Spark Streaming более уязвимым к ошибкам (особенно при попытке записать в нисходящие зависимости), и я не знаю, как лучше всего справляться с ошибками при попытке записать результаты в внешний источник, такой как Cassandra, DynamoDB и т.д.

Например, у меня есть задача Spark Streaming, которая извлекает данные из Stream (Kafka, Flume и т.д.), я еще не доработал, какую технологию использовать), объединяет похожие элементы вместе, а затем записывает результаты на внешний магазин. (т.е. Cassandra, DynamoDB или что-то другое, получающее результаты моих вычислений DStream).

Я пытаюсь понять, как я обрабатываю случай, когда внешняя зависимость недоступна для записи. Возможно, кластер пошел вниз, возможно, есть проблемы с разрешениями и т.д., Но моя работа не может записать результаты во внешнюю зависимость. Есть ли способ приостановить Spark Streaming, чтобы получатели не продолжали пакетные данные? Должен ли я просто спать в текущей партии и позволить Получателю продолжать хранить партии? Если проблема преходяща (несколько секунд), продолжение партии может быть приемлемым, но что произойдет, если зависимость снизится на несколько минут или 1 час (ы)?

Мне показалось, что у меня был монитор, который следит за состоянием зависимостей в фоновом режиме, и если он узнает, что он "нездоровый", он прекратит работу. Затем, когда все зависимости работоспособны, я могу запустить резервное копирование задания и обработать все данные, которые не были записаны во внешний источник.

Другая мысль, которую я имел, заключалась в том, чтобы каким-то образом сигнализировать в методе DStream forEachRdd, что была проблема. Есть ли какое-то исключение, которое я могу использовать в DStream, который будет сигнализировать драйверу, что он должен остановиться?

Если у кого-то есть опыт работы с функцией External Fault Tolerance, или может указать мне хорошие статьи/видеоролики, это было бы здорово.

Спасибо

Ответ 1

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

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

outputAction[T](rdd: RDD[T]): Unit = ???

по крайней мере убедитесь, что он не будет распространять исключение на ваш драйвер.

outputActionWithDelay[T](d: Duration)(rdd: RDD[T]) = ???

stream foreachRDD { rdd => Try(outputAction(rdd)) }

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

Это может быть дополнительно улучшено путем отслеживания сбоев и принятия некоторых других действий, если достигнут какой-либо порог.

Если данные сброса неприемлемы, следующий шаг - повторить попытку после некоторой задержки:

outputActionWithDelay[T](d: Duration)(rdd: RDD[T]) = ???

stream foreachRDD { 
  rdd => Try(outputAction(rdd))
    .recoverWith { case _ => Try(outputActionWithDelay(d1)(rdd)) }
    .recoverWith { case _ => Try(outputActionWithDelay(d2)(rdd)) }
   ...
}

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

Что вы можете сделать, когда ударяете последнюю попытку? Для начала мы можем добавить альтернативный источник вывода. Вместо использования первичного источника вы можете, например, выталкивать все в надежное хранилище внешних файлов и беспокоиться об этом позже. Это может неприменимо, если источник вывода требует определенного порядка входящих данных, но в противном случае стоит попробовать.

alternativeOutputAction[T](rdd: RDD[T]) = ???

stream foreachRDD { 
  rdd => Try(outputAction(rdd))
    .recoverWith { case _ => Try(outputActionWithDelay(d1)
    ...
    .recoverWith { case _ => Try(outputActionWithDelay(dn)(rdd)) }
    .recoverWith { case _ => Try(alternativeOutputAction(rdd))
}

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

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

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

Ответ 2

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