Расположение WAL в Spark структурированной потоковой передачи

Question

Я включил WAL для моего приложения структурированной потоковой передачи. Где я могу найти расположение журналов WAL? Я могу видеть WAL для моего процесса потоковой передачи Spark в префиксе receiveBlockMetadata . Но я не вижу никакого префикса, созданного для структурированной потоковой передачи

Answer 1

Согласно моему пониманию, WAL работает только в потоковом режиме с искрой, а не в структурированном режиме. Структурированная потоковая передача реализует отказоустойчивость, основанную на контрольной точке, например глобальное состояние flink. Контрольная точка хранит все состояния, включая смещения кафки и др. Местоположение указано в вашем коде.

Answer 2

Spark Streaming также имеет другую защиту от сбоев - журнал журналов, называемый Write Ahead Logs (WAL). Представленная в Spark 1.2, эта структура обеспечивает отказоустойчивость, сохраняя все данные, полученные получателями, в файл журнала, расположенный в каталоге checkpoint . Его можно включить с помощью свойства spark.streaming.receiver.writeAheadLog.enable.

Журналы сохраняются в receivedBlockMetadata/, расположенном в каталоге контрольных точек. К файлам добавляется лог. Как и в случае с контрольными точками, старые журналы автоматически очищаются Spark. Эту активность также можно наблюдать в журналах:

, например:

INFO Attempting to clear 0 old log files in file:/tmp/streaming_data_checkpoint/receivedBlockMetadata older than 1475022621700:  (org.apache.spark.streaming.util.FileBasedWriteAheadLog_ReceivedBlockTracker:54)

Пожалуйста, ознакомьтесь с этой книгой Orielly - Stream Streaming с Apache Spark

ИМХО, семантика отказоустойчивости одинакова для потоковой передачи Saprk и структурированной потоковой передачи Spark. См. Spark docs , подтверждающий это.

Настройка журналов упреждающей записи - Начиная с Spark 1.2, мы представили журналы опережающей записи для достижения надежных гарантий отказоустойчивости. Если этот параметр включен, все данные, полученные от получателя, записываются в журнал предварительной записи в каталоге контрольных точек конфигурации. Это предотвращает потерю данных при восстановлении драйвера, тем самым гарантируя нулевую потерю данных (подробно обсуждается в разделе «Семантика отказоустойчивости»). Это можно включить, установив для параметра конфигурации spark.streaming.receiver.writeAheadLog.enable значение true. Однако эта более сильная семантика может достигаться за счет пропускной способности приема отдельных приемников. Это можно исправить, запустив несколько приемников параллельно, чтобы увеличить совокупную пропускную способность. Кроме того, рекомендуется, чтобы репликация полученных данных в Spark была отключена, когда включен журнал записи с опережением, поскольку журнал уже хранится в реплицированной системе хранения. Это можно сделать, установив уровень хранения для входного потока в StorageLevel.MEMORY_AND_DISK_SER. При использовании S3 (или любой файловой системы, которая не поддерживает очистку) для журналов записи вперед, не забудьте включить spark.streaming.driver.writeAheadLog.closeFileAfterWrite и spark.streaming.receiver.writeAheadLog.closeFileAfterWrite. Смотрите Spark Streaming Configuration для более подробной информации. Обратите внимание, что Spark не будет шифровать данные, записанные в журнал записи с опережением, когда включено шифрование ввода-вывода. Если требуется шифрование данных журнала опережающей записи, они должны храниться в файловой системе, которая изначально поддерживает шифрование.

Надеюсь, как описано в документации, вы использовали эти 2 свойства как хорошо

spark.streaming.driver.writeAheadLog.closeFileAfterWrite
spark.streaming.receiver.writeAheadLog.closeFileAfterWrite