Проблемы с надежностью контрольных точек / WAL в Spark Streaming 1.6.0

Описание

У нас есть приложение Spark Streaming 1.5.2 на Scala, которое считывает события JSON из потока Kinesis, выполняет некоторые преобразования / агрегаты и записывает результаты в разные префиксы S3. Текущий пакетный интервал составляет 60 секунд. У нас 3000-7000 событий / сек. Мы используем контрольные точки, чтобы защитить себя от потери агрегированных данных.

Некоторое время он работает нормально, восстанавливается после исключений и даже перезапускается кластер. Недавно мы перекомпилировали код для Spark Streaming 1.6.0, изменив только зависимости библиотеки в файле build.sbt. После запуска кода в кластере Spark 1.6.0 в течение нескольких часов мы заметили следующее:

  1. Изменчивость «Скорость ввода» и «Время обработки» существенно увеличилась (см. Скриншоты ниже) в версии 1.6.0.
  2. Каждые несколько часов при записи записи BlockAdditionEvent… в WriteAheadLog возникает исключение «». java.util.concurrent.TimeoutException: время ожидания фьючерсов истекло после исключения [5000 миллисекунд] »(см. полную трассировку стека ниже), что совпадает со снижением до 0 событий в секунду для определенных пакетов (минут).

Покопавшись, я думаю, что вторая проблема, похоже, связана с этим запросом на слияние. Первоначальная цель PR: «При использовании S3 в качестве каталога для WAL запись занимает слишком много времени. Драйвер очень легко становится узким местом, когда несколько получателей отправляют события AddBlock в ReceiverTracker. Этот PR добавляет пакетирование событий в ReceivedBlockTracker, чтобы получатели не блокировались драйвером слишком долго ».

Мы устанавливаем контрольные точки в S3 в Spark 1.5.2, и проблем с производительностью / надежностью нет. Мы протестировали контрольную точку в Spark 1.6.0 на S3 и локальном NAS, и в обоих случаях мы получаем это исключение. Похоже, что когда для проверки пакета требуется более 5 секунд, возникает это исключение, и мы проверили, что события для этого пакета теряются навсегда.

Вопросов

  • Ожидается ли в Spark Streaming 1.6.0 увеличение волатильности «Скорость ввода» и «Время обработки» и есть ли какие-либо известные способы ее улучшения?

  • Знаете ли вы какое-либо обходное решение, кроме этих 2 ?:

    1) Чтобы гарантировать, что приемнику контрольной точки потребуется менее 5 секунд для записи всех файлов. По моему опыту, вы не можете гарантировать этого с S3 даже для небольших партий. Для локального NAS это зависит от того, кто отвечает за инфраструктуру (что сложно с поставщиками облачных услуг).

    2) Увеличьте значение свойства spark.streaming.driver.writeAheadLog.batchingTimeout.

  • Ожидаете ли вы проиграть какие-либо события в описанном сценарии? Я бы подумал, что если пакетная контрольная точка не удастся, порядковые номера осколка / получателя не будут увеличены, и это будет повторено позже.

Статистика Spark 1.5.2 - Скриншот

введите описание изображения здесь

Статистика Spark 1.6.0 - Скриншот

введите описание изображения здесь

Полная трассировка стека

16/01/19 03:25:03 WARN ReceivedBlockTracker: Exception thrown while writing record: BlockAdditionEvent(ReceivedBlockInfo(0,Some(3521),Some(SequenceNumberRanges(SequenceNumberRange(StreamEventsPRD,shardId-000000000003,49558087746891612304997255299934807015508295035511636018,49558087746891612304997255303224294170679701088606617650), SequenceNumberRange(StreamEventsPRD,shardId-000000000004,49558087949939897337618579003482122196174788079896232002,49558087949939897337618579006984380295598368799020023874), SequenceNumberRange(StreamEventsPRD,shardId-000000000001,49558087735072217349776025034858012188384702720257294354,49558087735072217349776025038332464993957147037082320914), SequenceNumberRange(StreamEventsPRD,shardId-000000000009,49558088270111696152922722880993488801473174525649617042,49558088270111696152922722884455852348849472550727581842), SequenceNumberRange(StreamEventsPRD,shardId-000000000000,49558087841379869711171505550483827793283335010434154498,49558087841379869711171505554030816148032657077741551618), SequenceNumberRange(StreamEventsPRD,shardId-000000000002,49558087853556076589569225785774419228345486684446523426,49558087853556076589569225789389107428993227916817989666))),BlockManagerBasedStoreResult(input-0-1453142312126,Some(3521)))) to the WriteAheadLog.
java.util.concurrent.TimeoutException: Futures timed out after [5000 milliseconds]
    at scala.concurrent.impl.Promise$DefaultPromise.ready(Promise.scala:219)
    at scala.concurrent.impl.Promise$DefaultPromise.result(Promise.scala:223)
    at scala.concurrent.Await$$anonfun$result$1.apply(package.scala:107)
    at scala.concurrent.BlockContext$DefaultBlockContext$.blockOn(BlockContext.scala:53)
    at scala.concurrent.Await$.result(package.scala:107)
    at org.apache.spark.streaming.util.BatchedWriteAheadLog.write(BatchedWriteAheadLog.scala:81)
    at org.apache.spark.streaming.scheduler.ReceivedBlockTracker.writeToLog(ReceivedBlockTracker.scala:232)
    at org.apache.spark.streaming.scheduler.ReceivedBlockTracker.addBlock(ReceivedBlockTracker.scala:87)
    at org.apache.spark.streaming.scheduler.ReceiverTracker.org$apache$spark$streaming$scheduler$ReceiverTracker$$addBlock(ReceiverTracker.scala:321)
    at org.apache.spark.streaming.scheduler.ReceiverTracker$ReceiverTrackerEndpoint$$anonfun$receiveAndReply$1$$anon$1$$anonfun$run$1.apply$mcV$sp(ReceiverTracker.scala:500)
    at org.apache.spark.util.Utils$.tryLogNonFatalError(Utils.scala:1230)
    at org.apache.spark.streaming.scheduler.ReceiverTracker$ReceiverTrackerEndpoint$$anonfun$receiveAndReply$1$$anon$1.run(ReceiverTracker.scala:498)
    at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142)
    at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617)
    at java.lang.Thread.run(Thread.java:745)

Исходный код

...
     // Function to create a new StreamingContext and set it up
  def setupContext(): StreamingContext = {
    ...
    // Create a StreamingContext
    val ssc = new StreamingContext(sc, Seconds(batchIntervalSeconds))

    // Create a Kinesis DStream
    val data = KinesisUtils.createStream(ssc,
      kinesisAppName, kinesisStreamName,
      kinesisEndpointUrl, RegionUtils.getRegionByEndpoint(kinesisEndpointUrl).getName(),
      InitialPositionInStream.LATEST, Seconds(kinesisCheckpointIntervalSeconds),
      StorageLevel.MEMORY_AND_DISK_SER_2, awsAccessKeyId, awsSecretKey)
...
    ssc.checkpoint(checkpointDir)

    ssc
  }


  // Get or create a streaming context.
  val ssc = StreamingContext.getActiveOrCreate(checkpointDir, setupContext)

  ssc.start()
  ssc.awaitTermination()

apache-spark spark-streaming scala amazon-kinesis checkpointing
person MiguelPeralvo    schedule 19.01.2016    source источник
comment
Интересно. Вы пытались уменьшить размер партии? Как защитить себя от повторной попытки продюсера?   -  person Leet-Falcon    schedule 13.02.2016
comment
Это помогло бы, если бы у нас не было некоторых ограничений вывода: мы пишем в S3, различные обрабатываемые RDD. При увеличении числа рабочих время обработки и вывода сокращается, но также увеличивается стоимость. Игра с соотношением разделов на одного рабочего, помогает. Настройка spark.streaming.blockInterval помогает, поскольку позволяет косвенно управлять количеством разделов, не вызывая перетасовки, которую вы получаете с помощью repartition () на уровне DStream. coalesce () на уровне RDD также помогает. Что вы имеете в виду под защитой от повторной попытки производителя? Обеспечение семантики одноразовой доставки на выходе для предотвращения дублирования?   -  person MiguelPeralvo    schedule 15.02.2016

Ответы (1)


arrow_upward
5
arrow_downward

Следуя предложению zero323 о публикации моего комментария в качестве ответа:

Увеличение spark.streaming.driver.writeAheadLog.batchingTimeout решило проблему тайм-аута контрольных точек. Мы сделали это после того, как убедились, что у нас есть для этого место. Мы уже некоторое время тестируем его. Поэтому я рекомендую увеличивать его только после тщательного рассмотрения.

ПОДРОБНОСТИ

Мы использовали эти 2 настройки в $ SPARK_HOME / conf / spark-defaults.conf:

spark.streaming.driver.writeAheadLog.allowBatching true spark.streaming.driver.writeAheadLog.batchingTimeout 15000

Изначально у нас было только значение spark.streaming.driver.writeAheadLog.allowBatching, равное true.

Перед изменением мы воспроизводили проблему, упомянутую в вопросе («... ReceivedBlockTracker: исключение, возникшее при записи записи ...») в тестовой среде. Это происходило каждые несколько часов. После изменения проблема исчезла. Мы запускали его несколько дней, прежде чем перейти к производству.

Мы обнаружили, что getBatchingTimeout () метод класса WriteAheadLogUtils имел значение по умолчанию 5000 мс, как показано здесь:

def getBatchingTimeout(conf: SparkConf): Long = {
    conf.getLong(DRIVER_WAL_BATCHING_TIMEOUT_CONF_KEY, defaultValue = 5000)
}
person MiguelPeralvo    schedule 29.07.2016