(version customization) lesson 12: Executor fault-tolerant security for Spark Streaming source code interpretation

The contents of this issue:

1. WAL fault-tolerant mechanism of Executor.

2. Message replay

The security fault tolerance of Executor is mainly data security fault tolerance, so why not consider the security fault tolerance of data computing?

The reason is that when calculating, Spark Streaming is fault-tolerant with the help of RDD on Spark Core, so it is naturally safe and reliable.

The security fault tolerance of Executor mainly includes:

1. Data copy:

There are two ways: a. Make a backup with the underlying BlockManager,BlockManager, and back up through the incoming StorageLevel.

B. Fault tolerance in WAL mode.

2. After receiving the data, no copy is made, but the data source supports storage. The so-called storage means that the source data can be read repeatedly.

The disadvantages of fault tolerance: time-consuming and space-consuming.

Take a simple look at the source code:

/ * * Store block and report it to driver * / def pushAndReportBlock (receivedBlock: ReceivedBlock,metadataOption: Option [Any], blockIdOption: Option [StreamBlockId]) {val blockId = blockIdOption.getOrElse (nextBlockId) val time = System.currentTimeMillisval blockStoreResult = receivedBlockHandler.storeBlock (blockId, receivedBlock) logDebug (s "Pushed block $blockId in ${(System.currentTimeMillis-time)} ms") val numRecords = blockStoreResult.numRecordsval blockInfo = ReceivedBlockInfo (streamId, numRecords, metadataOption, blockStoreResult) trackerEndpoint.askWithRetry [Boolean] (AddBlock (blockInfo)) logDebug (s "Reported block $Reported block")}

Privateval receivedBlockHandler: ReceivedBlockHandler = {if (WriteAheadLogUtils.enableReceiverLog (env.conf)) {if (checkpointDirOption.isEmpty) {throw new SparkException ("Cannot enable receiver write-ahead log without checkpoint directory set. + "Please use streamingContext.checkpoint () to set the checkpoint directory. "+" See documentation for more details. ")} new WriteAheadLogBasedBlockHandler (env.blockManager, receiver.streamId,receiver.storageLevel, env.conf, hadoopConf, checkpointDirOption.get) / / Fault tolerance via WAL} else {new BlockManagerBasedBlockHandler (env.blockManager, receiver.storageLevel) / / Fault tolerance via BlockManager}} def storeBlock (blockId: StreamBlockId, block: ReceivedBlock): ReceivedBlockStoreResult = {var numRecords = None: option [long] val putResult: Seq [(BlockId) BlockStatus)] = block match {case ArrayBufferBlock (arrayBuffer) = > numRecords = Some (arrayBuffer.size.toLong) blockManager.putIterator (blockId, arrayBuffer.iterator, storageLevel,tellMaster = true) case IteratorBlock (iterator) = > val countIterator = new CountingIterator (iterator) val putResult = blockManager.putIterator (blockId, countIterator, storageLevel,tellMaster = true) numRecords = countIterator.count putResultcase ByteBufferBlock (byteBuffer) = > blockManager.putBytes (blockId, byteBuffer, storageLevel,tellMaster = true) case o = > throw new SparkException (s "Could not store $blockId to block manager) Unexpected block type ${o.getClass.getName} ")} if (! putResult.map {_ .1} .targets (blockId)) {throw new SparkException (s" Could not store $blockId to block manager with storage level $storageLevel ")} BlockManagerBasedStoreResult (blockId, numRecords)}

Simple flowchart:

Reference blog: http://blog.csdn.net/hanburgud/article/details/51471089