This is an automated email from the ASF dual-hosted git repository.
cloud-fan pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/master by this push:
new bb76c7da9d7c [SPARK-58107][CORE] Detect and repair non-deterministic
local checkpoints via RDD-block content checksums
bb76c7da9d7c is described below
commit bb76c7da9d7c3f49baa594466fbca298705b929c
Author: Juliusz Sompolski <[email protected]>
AuthorDate: Wed Jul 15 11:51:19 2026 +0800
[SPARK-58107][CORE] Detect and repair non-deterministic local checkpoints
via RDD-block content checksums
### What changes were proposed in this pull request?
A locally-checkpointed RDD partition can be materialized by more than one
successful task attempt - from speculation, or a *zombie task* of a superseded
stage attempt during retries. When the computation is non-deterministic, the
attempts produce different bytes for the same `RDDBlockId(rddId, partIdx)`. The
BlockManager keeps both copies (the master's `blockLocations` simply appends
locations, with no version / attempt id) and a reader picks one via
`Random.shuffle`, so two reads of [...]
This adds a **dedup-and-seal** mechanism, enabled by default as a
correctness fix. A non-deterministic computation has no single correct value,
so it does not try to recompute one - it picks one surviving version per
partition and makes every later read agree on it. Three parts:
1. **Content checksum at store time.** When an RDD marked for verification
is materialized, the BlockManager folds a JDK checksum (CRC32C by default) over
the serialized + compressed *plaintext* of each cache block (composed above the
encrypting sink, so it is deterministic even under IO encryption) and reports
it per replica to the driver alongside the block location. The mark is plumbed
through `getOrElseUpdateRDDBlock -> getOrElseUpdate -> doPutIterator` and
recorded on the block's [...]
2. **Seal at the checkpoint commit point.**
`LocalRDDCheckpointData.doCheckpoint`, after materialization and before the
checkpoint is exposed to readers, seals each partition on the
`BlockManagerMasterEndpoint`: it picks one checksum value as authoritative,
evicts the replicas that disagree (dropping them from the directory and asking
their executors to drop the local copy, fire-and-forget), records the sealed
checksum, and rejects future divergent (or checksum-less) registrations. No
[...]
3. **Read-side self-check.** A read of a block on the seal path serves the
local copy only if its checksum equals the sealed value; otherwise it skips the
local copy and goes to an authoritative remote location - making correctness
independent of the asynchronous eviction landing.
The checksum is threaded across every path a block reaches the master
(store, eviction/spill, heartbeat re-report, and replication), so a sealed
block always reports its checksum and the master can safely reject a
checksum-less report. Five `.internal()` configs gate it, splitting "compute a
checksum" (`spark.storage.rddBlockChecksum.*`) from "seal a local checkpoint"
(`spark.checkpoint.local.verifyChecksum.*`). Only serialized storage levels
(e.g. `DISK_ONLY`) are covered; the uncove [...]
### Why are the changes needed?
Non-deterministic local checkpoints can silently produce inconsistent reads
across a partition, manifesting downstream as data-correctness failures
(row-count / invariant violations). Speculation can be disabled for
locally-checkpointed task sets, but retries are mandatory for fault tolerance,
so "just disable speculation" does not cover the retry / zombie-task case.
This particularly affects **Delta Lake's MERGE source materialization**,
which `localCheckpoint`s the source DataFrame at a serialized `DISK_ONLY` level
to cut lineage before MERGE's multiple jobs. A non-deterministic source
materialized inconsistently across those jobs surfaces as MERGE invariant /
row-count violations in the written table. This is a general Spark-core
correctness gap; Delta MERGE is the motivating downstream consumer.
### Does this PR introduce _any_ user-facing change?
Yes, as a correctness fix (all configs are `.internal()`). With the feature
on by default, a non-deterministic locally-checkpointed RDD now yields one
consistent version across all reads of a partition, instead of readers
potentially disagreeing. Deterministic checkpoints are unaffected. Setting
`spark.checkpoint.local.verifyChecksum.enabled=false` restores the prior
behavior.
### How was this patch tested?
New unit tests:
- `SerializerManagerSuite`: `wrapForChecksum` produces equal checksums for
identical bytes, different for divergent bytes.
- `LocalCheckpointSuite`: the verify mark is set only for serialized
checkpoints; `forceSerialized` bumps a default checkpoint to a serialized level.
- `BlockManagerSuite`: `sealRddChecksums` keeps the authoritative checksum
and evicts disagreeing replicas; a sealed block rejects a divergent (and a
checksum-less) registration and admits a matching one; the
checksumless-partition count is reported; sealed checksums are cleared on RDD
removal; agreeing disk and `_SER` replicas survive the seal; the
`UpdateBlockInfo` `Externalizable` round-trip preserves the checksum;
**divergence injection** - two executors materialize one value and [...]
A microbenchmark, `RddBlockChecksumBenchmark`, measures the store-time
overhead of the checksum layer (a few percent of serialize time uncompressed,
largion is on); committed results in
`core/benchmarks/RddBlockChecksumBenchmark-results.txt`.
A deterministic end-to-end test of scheduler-driven divergence (real
speculation, or a zombie task) is impractical - Spark kills redundant attempts
and aduler orchestration - so divergence is injected at the BlockManager level,
which exercises the same feature response (seal -> evict -> read-side skip).
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Claude Code Opus 4.8 (Anthropic)
Closes #57232 from juliuszsompolski/SPARK-58107-local-checkpoint-checksum.
Authored-by: Juliusz Sompolski <[email protected]>
Signed-off-by: Wenchen Fan <[email protected]>
---
.../RddBlockChecksumBenchmark-results.txt | 46 ++++
.../org/apache/spark/internal/config/package.scala | 69 +++++
.../apache/spark/network/BlockDataManager.scala | 8 +-
.../spark/network/BlockTransferService.scala | 13 +-
.../spark/network/netty/NettyBlockRpcServer.scala | 30 ++-
.../network/netty/NettyBlockTransferService.scala | 10 +-
.../apache/spark/rdd/LocalRDDCheckpointData.scala | 21 +-
core/src/main/scala/org/apache/spark/rdd/RDD.scala | 75 +++++-
.../spark/serializer/SerializerManager.scala | 23 +-
.../apache/spark/storage/BlockInfoManager.scala | 24 ++
.../org/apache/spark/storage/BlockManager.scala | 284 +++++++++++++++++---
.../apache/spark/storage/BlockManagerMaster.scala | 25 +-
.../spark/storage/BlockManagerMasterEndpoint.scala | 117 +++++++-
.../spark/storage/BlockManagerMessages.scala | 15 +-
.../apache/spark/storage/memory/MemoryStore.scala | 14 +-
.../spark/network/BlockTransferServiceSuite.scala | 4 +-
.../apache/spark/rdd/LocalCheckpointSuite.scala | 50 +++-
.../spark/serializer/SerializerManagerSuite.scala | 46 ++++
.../BlockManagerDecommissionUnitSuite.scala | 29 +-
.../apache/spark/storage/BlockManagerSuite.scala | 298 ++++++++++++++++++++-
.../spark/storage/FallbackStorageSuite.scala | 5 +-
.../spark/storage/RddBlockChecksumBenchmark.scala | 180 +++++++++++++
22 files changed, 1290 insertions(+), 96 deletions(-)
diff --git a/core/benchmarks/RddBlockChecksumBenchmark-results.txt
b/core/benchmarks/RddBlockChecksumBenchmark-results.txt
new file mode 100644
index 000000000000..6486252db032
--- /dev/null
+++ b/core/benchmarks/RddBlockChecksumBenchmark-results.txt
@@ -0,0 +1,46 @@
+================================================================================================
+RDD block store-time checksum overhead
+================================================================================================
+
+OpenJDK 64-Bit Server VM 17.0.19+10-LTS on Linux 5.4.0-1160-aws-fips
+Intel(R) Xeon(R) Platinum 8375C CPU @ 2.90GHz
+64k long rows, spark.rdd.compress=false: Best Time(ms) Avg Time(ms)
Stdev(ms) Rate(M/s) Per Row(ns) Relative time
+------------------------------------------------------------------------------------------------------------------------
+serialize only (feature off) 461 484
20 0.0 2303936.3 1.0X
+serialize + CRC32C checksum (feature on) 472 477
4 0.0 2361797.2 1.0X
+
+OpenJDK 64-Bit Server VM 17.0.19+10-LTS on Linux 5.4.0-1160-aws-fips
+Intel(R) Xeon(R) Platinum 8375C CPU @ 2.90GHz
+64k long rows, spark.rdd.compress=true: Best Time(ms) Avg Time(ms)
Stdev(ms) Rate(M/s) Per Row(ns) Relative time
+------------------------------------------------------------------------------------------------------------------------
+serialize only (feature off) 637 641
7 0.0 3186968.0 1.0X
+serialize + CRC32C checksum (feature on) 631 663
33 0.0 3153669.5 1.0X
+
+OpenJDK 64-Bit Server VM 17.0.19+10-LTS on Linux 5.4.0-1160-aws-fips
+Intel(R) Xeon(R) Platinum 8375C CPU @ 2.90GHz
+8k x 128B records, spark.rdd.compress=false: Best Time(ms) Avg Time(ms)
Stdev(ms) Rate(M/s) Per Row(ns) Relative time
+---------------------------------------------------------------------------------------------------------------------------
+serialize only (feature off) 192 197
3 0.0 959684.7 1.0X
+serialize + CRC32C checksum (feature on) 202 205
5 0.0 1008471.7 1.1X
+
+OpenJDK 64-Bit Server VM 17.0.19+10-LTS on Linux 5.4.0-1160-aws-fips
+Intel(R) Xeon(R) Platinum 8375C CPU @ 2.90GHz
+8k x 128B records, spark.rdd.compress=true: Best Time(ms) Avg Time(ms)
Stdev(ms) Rate(M/s) Per Row(ns) Relative time
+--------------------------------------------------------------------------------------------------------------------------
+serialize only (feature off) 290 292
2 0.0 1447887.3 1.0X
+serialize + CRC32C checksum (feature on) 294 295
1 0.0 1468980.5 1.0X
+
+
+================================================================================================
+Raw checksum algorithm over 4 MiB (reference)
+================================================================================================
+
+OpenJDK 64-Bit Server VM 17.0.19+10-LTS on Linux 5.4.0-1160-aws-fips
+Intel(R) Xeon(R) Platinum 8375C CPU @ 2.90GHz
+Checksum algorithm: Best Time(ms) Avg Time(ms)
Stdev(ms) Rate(M/s) Per Row(ns) Relative
+------------------------------------------------------------------------------------------------------------------------
+ADLER32 361 361
0 0.0 1409212.3 1.0X
+CRC32 33 33
0 0.0 129493.3 10.9X
+CRC32C 33 33
0 0.0 129364.8 10.9X
+
+
diff --git a/core/src/main/scala/org/apache/spark/internal/config/package.scala
b/core/src/main/scala/org/apache/spark/internal/config/package.scala
index c26eec184346..685ef6e11329 100644
--- a/core/src/main/scala/org/apache/spark/internal/config/package.scala
+++ b/core/src/main/scala/org/apache/spark/internal/config/package.scala
@@ -2965,6 +2965,75 @@ package object config {
.booleanConf
.createWithDefault(false)
+ private[spark] val STORAGE_RDD_BLOCK_CHECKSUM_ENABLED =
+ ConfigBuilder("spark.storage.rddBlockChecksum.enabled")
+ .internal()
+ .doc("When true, the BlockManager computes a content checksum over the
serialized bytes " +
+ "of every serialized RDD cache block at store time and reports it to
the driver. Only " +
+ "serialized blocks are covered; deserialized in-memory blocks are not
checksummed.")
+ .version("4.3.0")
+ .withBindingPolicy(ConfigBindingPolicy.NOT_APPLICABLE)
+ .booleanConf
+ .createWithDefault(false)
+
+ private[spark] val STORAGE_RDD_BLOCK_CHECKSUM_ALGORITHM =
+ ConfigBuilder("spark.storage.rddBlockChecksum.algorithm")
+ .internal()
+ .doc("The checksum algorithm used for RDD block content checksums (e.g.
local-checkpoint " +
+ "verification). Only built-in JDK algorithms are supported.")
+ .version("4.3.0")
+ .withBindingPolicy(ConfigBindingPolicy.NOT_APPLICABLE)
+ .stringConf
+ .transform(_.toUpperCase(Locale.ROOT))
+ .checkValues(Set("ADLER32", "CRC32", "CRC32C"))
+ .createWithDefault("CRC32C")
+
+ private[spark] val STORAGE_RDD_BLOCK_CHECKSUM_VERIFY_ON_REPLICATION =
+ ConfigBuilder("spark.storage.rddBlockChecksum.verifyOnReplication")
+ .internal()
+ .doc("When true, a replica of a checksummed RDD block recomputes the
content checksum over " +
+ "the received bytes to verify the transfer, instead of trusting the
checksum sent by the " +
+ "source. Off by default: the source's checksum is recorded directly,
since the transport " +
+ "layer already provides integrity.")
+ .version("4.3.0")
+ .withBindingPolicy(ConfigBindingPolicy.NOT_APPLICABLE)
+ .booleanConf
+ .createWithDefault(false)
+
+ private[spark] val LOCAL_CHECKPOINT_VERIFY_CHECKSUM_ENABLED =
+ ConfigBuilder("spark.checkpoint.local.verifyChecksum.enabled")
+ .internal()
+ .doc("When true, Spark fingerprints the serialized bytes of
locally-checkpointed RDD " +
+ "partitions at store time and, at the checkpoint commit point, detects
partitions that " +
+ "were materialized inconsistently by more than one task attempt (Spark
non-determinism " +
+ "combined with retries/speculation) and seals each partition to a
single version. Guards " +
+ "against silently inconsistent local checkpoints. This implies
checksum computation for " +
+ "the checkpointed RDD regardless of
spark.storage.rddBlockChecksum.enabled. Only applied " +
+ "to a localCheckpoint with a SERIALIZED storage level (e.g.
DISK_ONLY); a deserialized " +
+ "level (the default MEMORY_AND_DISK) has no checksummable bytes and is
left unverified - " +
+ "see spark.checkpoint.local.verifyChecksum.forceSerialized to opt a
default checkpoint " +
+ "into a serialized level. Sealing runs at checkpoint finalization: an
eager checkpoint " +
+ "is sealed before any consumer reads it, while a lazy one is sealed
after its first job " +
+ "materializes it (so reads within that job, before finalization, may
still see an " +
+ "unsealed copy).")
+ .version("4.3.0")
+ .withBindingPolicy(ConfigBindingPolicy.NOT_APPLICABLE)
+ .booleanConf
+ .createWithDefault(true)
+
+ private[spark] val LOCAL_CHECKPOINT_VERIFY_CHECKSUM_FORCE_SERIALIZED =
+ ConfigBuilder("spark.checkpoint.local.verifyChecksum.forceSerialized")
+ .internal()
+ .doc("When true (and verifyChecksum.enabled is true), localCheckpoint
adapts a " +
+ "deserialized storage level to its serialized equivalent so the
checkpoint's blocks " +
+ "can be checksummed and sealed. Off by default so localCheckpoint's
storage level is " +
+ "not silently changed; enable it to verify checkpoints that would
otherwise use the " +
+ "deserialized default.")
+ .version("4.3.0")
+ .withBindingPolicy(ConfigBindingPolicy.NOT_APPLICABLE)
+ .booleanConf
+ .createWithDefault(false)
+
private[spark] val STAGE_MAX_ATTEMPTS =
ConfigBuilder("spark.stage.maxAttempts")
.doc("Specify the max attempts for a stage - the spark job will be
aborted if any of its " +
diff --git
a/core/src/main/scala/org/apache/spark/network/BlockDataManager.scala
b/core/src/main/scala/org/apache/spark/network/BlockDataManager.scala
index 89177346a789..d4f227b5191f 100644
--- a/core/src/main/scala/org/apache/spark/network/BlockDataManager.scala
+++ b/core/src/main/scala/org/apache/spark/network/BlockDataManager.scala
@@ -63,7 +63,9 @@ trait BlockDataManager {
blockId: BlockId,
data: ManagedBuffer,
level: StorageLevel,
- classTag: ClassTag[_]): Boolean
+ classTag: ClassTag[_],
+ checksum: Option[Long] = None,
+ verifySealedChecksum: Boolean = false): Boolean
/**
* Put the given block that will be received as a stream.
@@ -74,7 +76,9 @@ trait BlockDataManager {
def putBlockDataAsStream(
blockId: BlockId,
level: StorageLevel,
- classTag: ClassTag[_]): StreamCallbackWithID
+ classTag: ClassTag[_],
+ checksum: Option[Long] = None,
+ verifySealedChecksum: Boolean = false): StreamCallbackWithID
/**
* Release locks acquired by [[putBlockData()]] and [[getLocalBlockData()]].
diff --git
a/core/src/main/scala/org/apache/spark/network/BlockTransferService.scala
b/core/src/main/scala/org/apache/spark/network/BlockTransferService.scala
index 635efc3e2262..160745f8efae 100644
--- a/core/src/main/scala/org/apache/spark/network/BlockTransferService.scala
+++ b/core/src/main/scala/org/apache/spark/network/BlockTransferService.scala
@@ -62,7 +62,11 @@ abstract class BlockTransferService extends BlockStoreClient
{
blockId: BlockId,
blockData: ManagedBuffer,
level: StorageLevel,
- classTag: ClassTag[_]): Future[Unit]
+ classTag: ClassTag[_],
+ // Source replica's content checksum and seal-path mark, so the receiver
can verify and record
+ // them (see BlockManager.BlockStoreUpdater.save).
+ checksum: Option[Long] = None,
+ verifySealedChecksum: Boolean = false): Future[Unit]
/**
* A special case of [[fetchBlocks]], as it fetches only one block and is
blocking.
@@ -117,8 +121,11 @@ abstract class BlockTransferService extends
BlockStoreClient {
blockId: BlockId,
blockData: ManagedBuffer,
level: StorageLevel,
- classTag: ClassTag[_]): Unit = {
- val future = uploadBlock(hostname, port, execId, blockId, blockData,
level, classTag)
+ classTag: ClassTag[_],
+ checksum: Option[Long] = None,
+ verifySealedChecksum: Boolean = false): Unit = {
+ val future = uploadBlock(hostname, port, execId, blockId, blockData,
level, classTag,
+ checksum, verifySealedChecksum)
ThreadUtils.awaitResult(future, Duration.Inf)
}
}
diff --git
a/core/src/main/scala/org/apache/spark/network/netty/NettyBlockRpcServer.scala
b/core/src/main/scala/org/apache/spark/network/netty/NettyBlockRpcServer.scala
index 03810292bd05..a7de0ffa83ed 100644
---
a/core/src/main/scala/org/apache/spark/network/netty/NettyBlockRpcServer.scala
+++
b/core/src/main/scala/org/apache/spark/network/netty/NettyBlockRpcServer.scala
@@ -121,13 +121,13 @@ class NettyBlockRpcServer(
new StreamHandle(streamId, numBlockIds).toByteBuffer)
case uploadBlock: UploadBlock =>
- // StorageLevel and ClassTag are serialized as bytes using our
JavaSerializer.
- val (level, classTag) = deserializeMetadata(uploadBlock.metadata)
+ val meta = deserializeMetadata(uploadBlock.metadata)
val data = new NioManagedBuffer(ByteBuffer.wrap(uploadBlock.blockData))
val blockId = BlockId(uploadBlock.blockId)
- logDebug(s"Receiving replicated block $blockId with level ${level} " +
+ logDebug(s"Receiving replicated block $blockId with level
${meta.level} " +
s"from ${client.getSocketAddress}")
- val blockStored = blockManager.putBlockData(blockId, data, level,
classTag)
+ val blockStored = blockManager.putBlockData(
+ blockId, data, meta.level, meta.classTag, meta.checksum,
meta.verifySealedChecksum)
if (blockStored) {
responseContext.onSuccess(ByteBuffer.allocate(0))
} else {
@@ -175,21 +175,33 @@ class NettyBlockRpcServer(
responseContext: RpcResponseCallback): StreamCallbackWithID = {
val message =
BlockTransferMessage.Decoder.fromByteBuffer(messageHeader).asInstanceOf[UploadBlockStream]
- val (level, classTag) = deserializeMetadata(message.metadata)
+ val meta = deserializeMetadata(message.metadata)
val blockId = BlockId(message.blockId)
- logDebug(s"Receiving replicated block $blockId with level ${level} as
stream " +
+ logDebug(s"Receiving replicated block $blockId with level ${meta.level} as
stream " +
s"from ${client.getSocketAddress}")
// This will return immediately, but will setup a callback on streamData
which will still
// do all the processing in the netty thread.
- blockManager.putBlockDataAsStream(blockId, level, classTag)
+ blockManager.putBlockDataAsStream(
+ blockId, meta.level, meta.classTag, meta.checksum,
meta.verifySealedChecksum)
}
- private def deserializeMetadata[T](metadata: Array[Byte]): (StorageLevel,
ClassTag[T]) = {
+ private def deserializeMetadata(metadata: Array[Byte]):
BlockReplicationMetadata = {
serializer
.newInstance()
.deserialize(ByteBuffer.wrap(metadata))
- .asInstanceOf[(StorageLevel, ClassTag[T])]
+ .asInstanceOf[BlockReplicationMetadata]
}
override def getStreamManager(): StreamManager = streamManager
}
+
+/**
+ * Metadata in the (Java-serialized) `metadata` field of an
`UploadBlock`/`UploadBlockStream`.
+ * `checksum` and `verifySealedChecksum` carry a source replica's RDD-block
content checksum and its
+ * local-checkpoint seal-path mark (separate fields so the checksum is usable
without sealing).
+ */
+private[spark] case class BlockReplicationMetadata(
+ level: StorageLevel,
+ classTag: ClassTag[_],
+ checksum: Option[Long] = None,
+ verifySealedChecksum: Boolean = false)
diff --git
a/core/src/main/scala/org/apache/spark/network/netty/NettyBlockTransferService.scala
b/core/src/main/scala/org/apache/spark/network/netty/NettyBlockTransferService.scala
index fe55518d8000..9eaebe723b16 100644
---
a/core/src/main/scala/org/apache/spark/network/netty/NettyBlockTransferService.scala
+++
b/core/src/main/scala/org/apache/spark/network/netty/NettyBlockTransferService.scala
@@ -174,13 +174,15 @@ private[spark] class NettyBlockTransferService(
blockId: BlockId,
blockData: ManagedBuffer,
level: StorageLevel,
- classTag: ClassTag[_]): Future[Unit] = {
+ classTag: ClassTag[_],
+ checksum: Option[Long] = None,
+ verifySealedChecksum: Boolean = false): Future[Unit] = {
val result = Promise[Unit]()
val client = clientFactory.createClient(hostname, port)
- // StorageLevel and ClassTag are serialized as bytes using our
JavaSerializer.
- // Everything else is encoded using our binary protocol.
- val metadata =
JavaUtils.bufferToArray(serializer.newInstance().serialize((level, classTag)))
+ // BlockReplicationMetadata is Java-serialized; everything else uses our
binary protocol.
+ val metadata = JavaUtils.bufferToArray(serializer.newInstance().serialize(
+ BlockReplicationMetadata(level, classTag, checksum,
verifySealedChecksum)))
// We always transfer shuffle blocks as a stream for simplicity with the
receiving code since
// they are always written to disk. Otherwise we check the block size.
diff --git
a/core/src/main/scala/org/apache/spark/rdd/LocalRDDCheckpointData.scala
b/core/src/main/scala/org/apache/spark/rdd/LocalRDDCheckpointData.scala
index f9b2ffc068b0..ce21d45290e4 100644
--- a/core/src/main/scala/org/apache/spark/rdd/LocalRDDCheckpointData.scala
+++ b/core/src/main/scala/org/apache/spark/rdd/LocalRDDCheckpointData.scala
@@ -55,6 +55,14 @@ private[spark] class LocalRDDCheckpointData[T:
ClassTag](@transient private val
rdd.sparkContext.runJob(rdd, action,
missingPartitionIndices.toImmutableArraySeq)
}
+ // Finalization point: partitions are materialized (missing ones by the
runJob above) and the
+ // checkpoint is not yet exposed to readers (markCheckpointed runs after
doCheckpoint returns),
+ // so seal one consistent version per partition here. See
`RDD.sealCheckpointChecksums` for the
+ // timing guarantees (eager vs lazy).
+ if (rdd.verifyCheckpointChecksums) {
+ rdd.sealCheckpointChecksums()
+ }
+
new LocalCheckpointRDD[T](rdd)
}
@@ -71,9 +79,18 @@ private[spark] object LocalRDDCheckpointData {
* executors do not fail. Otherwise, if the RDD is cached in memory only,
for instance,
* the checkpoint data will be lost if the relevant block is evicted from
memory.
*
+ * When `forceSerialized` is set, the level is additionally adapted to a
serialized one, so a
+ * checkpoint whose blocks would otherwise be deserialized (and thus have no
bytes to
+ * content-checksum) becomes verifiable (see
`LOCAL_CHECKPOINT_VERIFY_CHECKSUM_FORCE_SERIALIZED`).
+ *
* This method is idempotent.
*/
- def transformStorageLevel(level: StorageLevel): StorageLevel = {
- StorageLevel(useDisk = true, level.useMemory, level.deserialized,
level.replication)
+ def transformStorageLevel(
+ level: StorageLevel, forceSerialized: Boolean = false): StorageLevel = {
+ StorageLevel(
+ useDisk = true,
+ useMemory = level.useMemory,
+ deserialized = level.deserialized && !forceSerialized,
+ replication = level.replication)
}
}
diff --git a/core/src/main/scala/org/apache/spark/rdd/RDD.scala
b/core/src/main/scala/org/apache/spark/rdd/RDD.scala
index 5500f085de1e..e4cbbe6302af 100644
--- a/core/src/main/scala/org/apache/spark/rdd/RDD.scala
+++ b/core/src/main/scala/org/apache/spark/rdd/RDD.scala
@@ -375,6 +375,55 @@ abstract class RDD[T: ClassTag](
}
}
+ /**
+ * Set (when verification is enabled) to mark this RDD's cache blocks for
content-checksum + seal.
+ * Read in `getOrCompute`, it drives checksum computation at store time and
is recorded on the
+ * executor-local `BlockInfo` so the read-side self-check enforces the seal
on that block; at the
+ * checkpoint commit point it decides whether to seal. Travels with the RDD
to executors because
+ * they need to know in advance - before storing/reading a block - whether
to compute its checksum
+ * and later apply the read-side sealed-checksum self-check.
+ */
+ private[rdd] var verifyCheckpointChecksums: Boolean = false
+
+ /**
+ * Seal this RDD's checksummed blocks so every later read sees a single
consistent version, even
+ * if Spark non-determinism plus speculation or stage retries materialized a
partition into more
+ * than one divergent copy. The master picks one checksum value per
partition, evicts the copies
+ * that disagree with it, and rejects future divergent registrations; reads
then self-check
+ * against the sealed checksum. No-op unless this RDD is marked for
verification. Relies on the
+ * per-replica checksums recorded by BlockManager at store time (see
+ * `SerializerManager.wrapForChecksum`).
+ *
+ * Called from `doCheckpoint`, after its `runJob` has materialized every
partition (so all copies
+ * are registered by seal time) and just before lineage is cut. Because
lineage is cut right
+ * after, there can be no later recompute producing a fresh,
differently-`rddId`'d version to
+ * reconcile: the sealed copies are the only ones there will ever be, and if
they are lost they
+ * are lost forever. With an eager checkpoint the RDD is fully materialized
before any consumer
+ * reads it, so every read sees the sealed version. With a lazy checkpoint
the seal still runs
+ * after the first job materializes the RDD, but reads *within that same
job*, before
+ * finalization, may still observe an unsealed (possibly divergent) copy;
only reads after
+ * finalization are guaranteed consistent.
+ *
+ * `verifyCheckpointChecksums` must be set before materialization starts;
otherwise partitions
+ * materialized while it was still unset carry no checksum, cannot be
sealed, and are reported by
+ * the warning below.
+ */
+ private[rdd] def sealCheckpointChecksums(): Unit = {
+ if (!verifyCheckpointChecksums) {
+ logWarning(log"sealCheckpointChecksums called on RDD ${MDC(RDD_ID, id)}
that was not " +
+ log"marked for checksum verification; nothing to seal.")
+ return
+ }
+ val unverified = SparkEnv.get.blockManager.master.sealRddChecksums(id)
+ if (unverified > 0) {
+ // A partition with no checksum was materialized before this RDD was
marked for verification
+ // (e.g. a prior persist() + action), or under a deserialized storage
level.
+ logWarning(log"Content verification is enabled for RDD ${MDC(RDD_ID,
id)} but " +
+ log"${MDC(NUM_PARTITIONS, unverified)} partition(s) were materialized
before it was " +
+ log"marked and were left unverified.")
+ }
+ }
+
/**
* Gets or computes an RDD partition. Used by RDD.iterator() when an RDD is
cached.
*/
@@ -386,7 +435,8 @@ abstract class RDD[T: ClassTag](
context.taskAttemptId(), blockId, storageLevel, elementClassTag, () => {
readCachedBlock = false
computeOrReadCheckpoint(partition, context)
- }
+ },
+ verifySealedChecksum = verifyCheckpointChecksums
) match {
// Block hit.
case Left(blockResult) =>
@@ -1737,11 +1787,20 @@ abstract class RDD[T: ClassTag](
// the storage level he/she specified to one that is appropriate for local
checkpointing
// (i.e. uses disk) to guarantee correctness.
- if (storageLevel == StorageLevel.NONE) {
- persist(LocalRDDCheckpointData.DEFAULT_STORAGE_LEVEL)
+ // Content verification only covers serialized blocks. When enabled
together with the
+ // force-serialized flag, adapt the level to a serialized one so a plain
localCheckpoint()
+ // (whose default level is deserialized) becomes verifiable.
+ val verifyCheckpointChecksumEnabled =
conf.get(LOCAL_CHECKPOINT_VERIFY_CHECKSUM_ENABLED)
+ val forceCheckpointSerialized =
+ verifyCheckpointChecksumEnabled &&
+ conf.get(LOCAL_CHECKPOINT_VERIFY_CHECKSUM_FORCE_SERIALIZED)
+ val checkpointLevel = if (storageLevel == StorageLevel.NONE) {
+ LocalRDDCheckpointData.transformStorageLevel(
+ LocalRDDCheckpointData.DEFAULT_STORAGE_LEVEL,
forceCheckpointSerialized)
} else {
- persist(LocalRDDCheckpointData.transformStorageLevel(storageLevel),
allowOverride = true)
+ LocalRDDCheckpointData.transformStorageLevel(storageLevel,
forceCheckpointSerialized)
}
+ persist(checkpointLevel, allowOverride = true)
// If this RDD is already checkpointed and materialized, its lineage is
already truncated.
// We must not override our `checkpointData` in this case because it is
needed to recover
@@ -1758,6 +1817,14 @@ abstract class RDD[T: ClassTag](
case _ =>
}
checkpointData = Some(new LocalRDDCheckpointData(this))
+ // Mark for checksum + seal only when the checkpoint's storage level is
serialized: a
+ // deserialized level keeps in-memory objects with no bytes to checksum,
so there is
+ // nothing to verify and marking would only add cost. (A deserialized
default is expected,
+ // hence no warning; `...forceSerialized` above opts a default
checkpoint into a serialized
+ // level so it can be verified.) `getStorageLevel` reflects the level
set above.
+ if (verifyCheckpointChecksumEnabled && !getStorageLevel.deserialized) {
+ verifyCheckpointChecksums = true
+ }
}
this
}
diff --git
a/core/src/main/scala/org/apache/spark/serializer/SerializerManager.scala
b/core/src/main/scala/org/apache/spark/serializer/SerializerManager.scala
index d53a4d549782..2370678ff5a6 100644
--- a/core/src/main/scala/org/apache/spark/serializer/SerializerManager.scala
+++ b/core/src/main/scala/org/apache/spark/serializer/SerializerManager.scala
@@ -19,12 +19,13 @@ package org.apache.spark.serializer
import java.io.{BufferedInputStream, BufferedOutputStream, InputStream,
OutputStream}
import java.nio.ByteBuffer
+import java.util.zip.Checksum
import scala.reflect.ClassTag
import org.apache.spark.SparkConf
import org.apache.spark.internal.config
-import org.apache.spark.io.CompressionCodec
+import org.apache.spark.io.{CompressionCodec, MutableCheckedOutputStream}
import org.apache.spark.security.CryptoStreamUtils
import org.apache.spark.storage._
import org.apache.spark.util.io.{ChunkedByteBuffer,
ChunkedByteBufferOutputStream}
@@ -159,6 +160,26 @@ private[spark] class SerializerManager(
if (shouldCompress(blockId)) compressionCodec.compressedOutputStream(s)
else s
}
+ /**
+ * Wrap an output stream so that bytes written through it are folded into
`checksum`.
+ *
+ * Compose this INSIDE `wrapForCompression` and ABOVE the sink, i.e.
+ * `ser.serializeStream(wrapForCompression(blockId,
wrapForChecksum(checksum, sink)))`, so the
+ * checksum observes the serialized+compressed *plaintext* rather than the
possibly-encrypted
+ * on-disk bytes (DiskStore applies encryption below the sink with a random
IV). That makes the
+ * fingerprint deterministic and identical across storage representations
and replicas, so a
+ * consumer can tell whether two replicas of a block hold the same bytes.
+ */
+ def wrapForChecksum(checksum: Checksum, s: OutputStream): OutputStream = {
+ val checked = new MutableCheckedOutputStream(s)
+ checked.setChecksum(checksum)
+ checked
+ }
+
+ /** Wrap `s` for `checksum` when present, otherwise return it unchanged. */
+ def wrapForChecksum(checksum: Option[Checksum], s: OutputStream):
OutputStream =
+ checksum.map(wrapForChecksum(_, s)).getOrElse(s)
+
/**
* Wrap an input stream for compression if block compression is enabled for
its block type
*/
diff --git
a/core/src/main/scala/org/apache/spark/storage/BlockInfoManager.scala
b/core/src/main/scala/org/apache/spark/storage/BlockInfoManager.scala
index 417705e40599..4dc8d1f1b50c 100644
--- a/core/src/main/scala/org/apache/spark/storage/BlockInfoManager.scala
+++ b/core/src/main/scala/org/apache/spark/storage/BlockInfoManager.scala
@@ -83,6 +83,30 @@ private[storage] class BlockInfo(
}
private[this] var _writerTask: Long = BlockInfo.NO_WRITER
+ // The three checksum fields below are `@volatile`: unlike the lock-guarded
fields above, they are
+ // read/written across threads outside the block lock (e.g. a lockless
`reportAllBlocks` heartbeat
+ // read, and a write under only a shared read lock in
`localCopyMatchesSealedChecksum`), so they
+ // need their own visibility guarantee.
+
+ /**
+ * Content checksum of this block's serialized+compressed (pre-encryption)
bytes, computed at
+ * store time when a checksum was requested for it. `None` when no checksum
was computed.
+ */
+ @volatile var checksum: Option[Long] = None
+
+ /**
+ * Whether this block is subject to the read-side seal self-check (see
`getLocalValues`). Set for
+ * blocks of RDDs with `verifyCheckpointChecksums` set; a block merely
checksummed for observation
+ * is not.
+ */
+ @volatile var verifySealedChecksum: Boolean = false
+
+ /**
+ * Cached sealed checksum for this block, pulled from the master once on
first read and reused.
+ * `None` until pulled, or if the block is not sealed.
+ */
+ @volatile var sealedChecksum: Option[Long] = None
+
private def checkInvariants(): Unit = {
// A block's reader count must be non-negative:
assert(_readerCount >= 0)
diff --git a/core/src/main/scala/org/apache/spark/storage/BlockManager.scala
b/core/src/main/scala/org/apache/spark/storage/BlockManager.scala
index 5fbc8dca74f6..21f932ba057a 100644
--- a/core/src/main/scala/org/apache/spark/storage/BlockManager.scala
+++ b/core/src/main/scala/org/apache/spark/storage/BlockManager.scala
@@ -23,6 +23,7 @@ import java.nio.ByteBuffer
import java.nio.channels.Channels
import java.util.Collections
import java.util.concurrent.{CompletableFuture, ConcurrentHashMap, TimeUnit}
+import java.util.zip.Checksum
import scala.collection.mutable
import scala.collection.mutable.HashMap
@@ -230,6 +231,83 @@ private[spark] class BlockManager(
/** Whether rdd cache visibility tracking is enabled. */
private val trackingCacheVisibility: Boolean =
conf.get(RDD_CACHE_VISIBILITY_TRACKING_ENABLED)
+ /** Whether to compute a content checksum for every serialized RDD cache
block at store time. */
+ private val rddBlockChecksumEnabled: Boolean =
+ conf.get(config.STORAGE_RDD_BLOCK_CHECKSUM_ENABLED)
+
+ /** Algorithm used for RDD block checksums (a built-in JDK checksum, shared
with shuffle). */
+ private val rddBlockChecksumAlgorithm: String =
+ conf.get(config.STORAGE_RDD_BLOCK_CHECKSUM_ALGORITHM)
+
+ /** Whether a replica recomputes the checksum on receive rather than
trusting the sent value. */
+ private val rddBlockChecksumVerifyOnReplication: Boolean =
+ conf.get(config.STORAGE_RDD_BLOCK_CHECKSUM_VERIFY_ON_REPLICATION)
+
+ /**
+ * Returns a fresh content-checksum accumulator for an RDD block, or None
when checksum
+ * computation is disabled or this is not an RDD block.
+ *
+ * Callers wrap their serialization sink with
`SerializerManager.wrapForChecksum` so the checksum
+ * covers the serialized+compressed (pre-encryption) bytes - the
deterministic,
+ * representation-independent form, identical across the disk and in-memory
paths and across
+ * replicas - then record `checksum.getValue` on the block's `BlockInfo`
once it is fully
+ * written. The driver tracks it per replica (`BlockManagerMasterEndpoint`),
which lets a consumer
+ * detect a block produced inconsistently by more than one task attempt
(Spark non-determinism
+ * together with speculation or stage retries).
+ */
+ private def newRddBlockChecksum(
+ blockId: BlockId, verifySealedChecksum: Boolean): Option[Checksum] = {
+ // Compute when the global switch is on, or when this block will be sealed.
+ if ((rddBlockChecksumEnabled || verifySealedChecksum) && blockId.isRDD) {
+
Some(ShuffleChecksumHelper.getChecksumByAlgorithm(rddBlockChecksumAlgorithm))
+ } else {
+ None
+ }
+ }
+
+ /**
+ * Fold a content checksum over a received (replicated) RDD block's bytes.
The bytes are the
+ * decrypted serialized+compressed plaintext (the same form
`wrapForChecksum` covers on the store
+ * path), so the value is comparable to the source's and to the seal.
+ */
+ private def computeReceivedBlockChecksum(data: BlockData): Long = {
+ val checksum =
ShuffleChecksumHelper.getChecksumByAlgorithm(rddBlockChecksumAlgorithm)
+ val out = serializerManager.wrapForChecksum(checksum,
OutputStream.nullOutputStream())
+ Utils.tryWithResource(data.toInputStream()) { in => in.transferTo(out) }
+ out.close()
+ checksum.getValue
+ }
+
+ /**
+ * For a block that carries its own content checksum, whether this
executor's local copy is the
+ * one to serve: it must match the block's sealed checksum. The sealed value
is pulled from the
+ * master once and cached on the `BlockInfo` (immutable once set). A block
with no sealed checksum
+ * yet imposes no constraint - the local copy is served as usual; only once
a block is sealed
+ * does a non-matching local copy become a stale replica to skip in favor of
a remote one.
+ */
+ private def localCopyMatchesSealedChecksum(blockId: BlockId, info:
BlockInfo): Boolean = {
+ // The sealed value is cached once known (immutable thereafter). While
still unsealed we re-pull
+ // on each read rather than caching a "not sealed" sentinel: the seal is
applied on the driver
+ // and can land between two reads of the same local block, so a cached
"unsealed" would let a
+ // later read serve a now-divergent copy without checking - a correctness
hole. Resolving once
+ // up front (e.g. at `getOrElseUpdateRDDBlock` entry, as visibility
tracking does) does not help
+ // either: the block is typically produced and read back within that same
call, before the seal
+ // exists. This costs a driver round-trip per pre-seal read, but the only
pre-seal reader is the
+ // producing job itself (the eager store-time readback), so the window is
bounded.
+ if (info.sealedChecksum.isEmpty) {
+ info.sealedChecksum = master.getSealedChecksum(blockId)
+ }
+ info.sealedChecksum.isEmpty || info.sealedChecksum == info.checksum
+ }
+
+ /**
+ * Record a freshly-computed block checksum on its `BlockInfo` once the
serialized bytes have been
+ * fully written. Pair every store-path `newRddBlockChecksum` +
`wrapForChecksum` with this so the
+ * folded value reaches the master; a no-op when no checksum was computed.
+ */
+ private def recordChecksum(info: BlockInfo, checksum: Option[Checksum]):
Unit =
+ checksum.foreach(c => info.checksum = Some(c.getValue))
+
// Visible for testing
private[storage] val blockInfoManager = new
BlockInfoManager(trackingCacheVisibility)
@@ -375,6 +453,10 @@ private[spark] class BlockManager(
* Abstraction for storing blocks from bytes, whether they start in memory
or on disk.
*
* @param blockSize the decrypted size of the block
+ * @param sourceChecksum the source replica's content checksum, when it had
one (seal path or the
+ * global compute switch); used to verify the transfer
(see `save`).
+ * @param sourceVerifySealedChecksum whether the source was on the seal
path, propagated so this
+ * replica records the mark and self-checks reads.
*/
private[spark] abstract class BlockStoreUpdater[T](
blockSize: Long,
@@ -382,7 +464,9 @@ private[spark] class BlockManager(
level: StorageLevel,
classTag: ClassTag[T],
tellMaster: Boolean,
- keepReadLock: Boolean) {
+ keepReadLock: Boolean,
+ sourceChecksum: Option[Long] = None,
+ sourceVerifySealedChecksum: Boolean = false) {
/**
* Reads the block content into the memory. If the update of the block
store is based on a
@@ -446,12 +530,32 @@ private[spark] class BlockManager(
val replicationFuture = if (level.replication > 1) {
Future {
// This is a blocking action and should run in
futureExecutionContext which is a cached
- // thread pool.
- replicate(blockId, blockData(), level, classTag)
+ // thread pool. Forward this replica's checksum + seal mark so
downstream replicas are
+ // verified against the seal too.
+ replicate(blockId, blockData(), level, classTag,
+ sourceChecksum = info.checksum,
+ sourceVerifySealedChecksum = info.verifySealedChecksum)
}(futureExecutionContext)
} else {
null
}
+ // With verifyOnReplication on, recompute the checksum over the
received bytes to verify the
+ // transfer. Do it here, before saveToDiskStore() runs: on the
stream-upload path
+ // (TempFileBasedBlockStoreUpdater) that store moves the temp file
away, so blockData() is
+ // no longer readable afterward. Recorded on info.checksum only once
the store succeeds.
+ val recomputedChecksum =
+ if (blockId.isRDD && rddBlockChecksumVerifyOnReplication &&
+ (sourceChecksum.isDefined || sourceVerifySealedChecksum)) {
+ val recomputed = computeReceivedBlockChecksum(blockData())
+ if (sourceChecksum.exists(_ != recomputed)) {
+ logWarning(log"Replicated block ${MDC(BLOCK_ID, blockId)}
arrived with a source " +
+ log"checksum that does not match its received bytes; storing
the recomputed " +
+ log"value (possible transmission corruption or
non-deterministic serialization).")
+ }
+ Some(recomputed)
+ } else {
+ None
+ }
if (level.useMemory) {
// Put it in memory first, even if it also has useDisk set to true;
// We will drop it to disk later if the memory store can't hold it.
@@ -470,11 +574,20 @@ private[spark] class BlockManager(
val putBlockStatus = getCurrentBlockStatus(blockId, info)
val blockWasSuccessfullyStored = putBlockStatus.storageLevel.isValid
if (blockWasSuccessfullyStored) {
+ // Propagate the source replica's verify mark and record its content
checksum, so the
+ // master can verify this replica and this executor's reads
self-check it. By default the
+ // source's checksum is trusted; with verifyOnReplication on, the
recompute over the
+ // received bytes (done before the store above) also verifies the
transfer.
+ if (blockId.isRDD) {
+ if (sourceVerifySealedChecksum) info.verifySealedChecksum = true
+ // Prefer the recompute (verifyOnReplication) over the trusted
source checksum.
+ info.checksum = recomputedChecksum.orElse(sourceChecksum)
+ }
// Now that the block is in either the memory or disk store,
// tell the master about it.
info.size = blockSize
if (tellMaster && info.tellMaster) {
- reportBlockStatus(blockId, putBlockStatus)
+ reportBlockStatus(blockId, putBlockStatus, checksum =
info.checksum)
}
addUpdatedBlockStatusToTaskMetrics(blockId, putBlockStatus)
}
@@ -507,8 +620,11 @@ private[spark] class BlockManager(
classTag: ClassTag[T],
bytes: ChunkedByteBuffer,
tellMaster: Boolean = true,
- keepReadLock: Boolean = false)
- extends BlockStoreUpdater[T](bytes.size, blockId, level, classTag,
tellMaster, keepReadLock) {
+ keepReadLock: Boolean = false,
+ sourceChecksum: Option[Long] = None,
+ sourceVerifySealedChecksum: Boolean = false)
+ extends BlockStoreUpdater[T](bytes.size, blockId, level, classTag,
tellMaster, keepReadLock,
+ sourceChecksum, sourceVerifySealedChecksum) {
override def readToByteBuffer(): ChunkedByteBuffer = bytes
@@ -532,8 +648,11 @@ private[spark] class BlockManager(
tmpFile: File,
blockSize: Long,
tellMaster: Boolean = true,
- keepReadLock: Boolean = false)
- extends BlockStoreUpdater[T](blockSize, blockId, level, classTag,
tellMaster, keepReadLock) {
+ keepReadLock: Boolean = false,
+ sourceChecksum: Option[Long] = None,
+ sourceVerifySealedChecksum: Boolean = false)
+ extends BlockStoreUpdater[T](blockSize, blockId, level, classTag,
tellMaster, keepReadLock,
+ sourceChecksum, sourceVerifySealedChecksum) {
override def readToByteBuffer(): ChunkedByteBuffer = {
val allocator = level.memoryMode match {
@@ -687,7 +806,10 @@ private[spark] class BlockManager(
logInfo(log"Reporting ${MDC(NUM_BLOCKS, blockInfoManager.size)} blocks to
the master.")
for ((blockId, info) <- blockInfoManager.entries) {
val status = getCurrentBlockStatus(blockId, info)
- if (info.tellMaster && !tryToReportBlockStatus(blockId, status)) {
+ // Forward the block's content checksum so a re-report re-verifies
against the seal (a sealed
+ // block must never report checksum-less; see
BlockManagerMasterEndpoint.updateBlockInfo).
+ if (info.tellMaster &&
+ !tryToReportBlockStatus(blockId, status, checksum = info.checksum)) {
logError(log"Failed to report ${MDC(BLOCK_ID, blockId)} to master;
giving up.")
return
}
@@ -793,14 +915,19 @@ private[spark] class BlockManager(
blockId: BlockId,
data: ManagedBuffer,
level: StorageLevel,
- classTag: ClassTag[_]): Boolean = {
- putBytes(blockId, new ChunkedByteBuffer(data.nioByteBuffer()),
level)(classTag)
+ classTag: ClassTag[_],
+ checksum: Option[Long] = None,
+ verifySealedChecksum: Boolean = false): Boolean = {
+ putBytes(blockId, new ChunkedByteBuffer(data.nioByteBuffer()), level,
+ sourceChecksum = checksum, sourceVerifySealedChecksum =
verifySealedChecksum)(classTag)
}
override def putBlockDataAsStream(
blockId: BlockId,
level: StorageLevel,
- classTag: ClassTag[_]): StreamCallbackWithID = {
+ classTag: ClassTag[_],
+ checksum: Option[Long] = None,
+ verifySealedChecksum: Boolean = false): StreamCallbackWithID = {
checkShouldStore(blockId, level)
@@ -837,7 +964,9 @@ private[spark] class BlockManager(
channel.close()
val blockSize = channel.getCount
val blockStored = TempFileBasedBlockStoreUpdater(
- blockId, level, classTag, tmpFile, blockSize).save()
+ blockId, level, classTag, tmpFile, blockSize,
+ sourceChecksum = checksum,
+ sourceVerifySealedChecksum = verifySealedChecksum).save()
if (!blockStored) {
throw
SparkCoreErrors.failToStoreBlockOnBlockManagerError(blockManagerId, blockId)
}
@@ -910,8 +1039,10 @@ private[spark] class BlockManager(
private[spark] def reportBlockStatus(
blockId: BlockId,
status: BlockStatus,
- droppedMemorySize: Long = 0L): Unit = {
- val needReregister = !tryToReportBlockStatus(blockId, status,
droppedMemorySize)
+ droppedMemorySize: Long = 0L,
+ checksum: Option[Long] = None): Unit = {
+ val needReregister =
+ !tryToReportBlockStatus(blockId, status, droppedMemorySize, checksum)
if (needReregister) {
logInfo(log"Got told to re-register updating block ${MDC(BLOCK_ID,
blockId)}")
// Re-registering will report our new block for free.
@@ -928,14 +1059,15 @@ private[spark] class BlockManager(
private def tryToReportBlockStatus(
blockId: BlockId,
status: BlockStatus,
- droppedMemorySize: Long = 0L): Boolean = {
+ droppedMemorySize: Long = 0L,
+ checksum: Option[Long] = None): Boolean = {
val storageLevel = status.storageLevel
val inMemSize = Math.max(status.memSize, droppedMemorySize)
val onDiskSize = status.diskSize
// Yet `blockId` could only be `ShuffleIndexBlockId` or
`ShuffleDataBlockId` when it's a
// shuffle block because of decommission.
val bmId = if (blockId.isShuffle) shuffleServerId else blockManagerId
- master.updateBlockInfo(bmId, blockId, storageLevel, inMemSize, onDiskSize)
+ master.updateBlockInfo(bmId, blockId, storageLevel, inMemSize, onDiskSize,
checksum)
}
/**
@@ -994,13 +1126,36 @@ private[spark] class BlockManager(
/**
* Get block from local block manager as an iterator of Java objects.
*/
- def getLocalValues(blockId: BlockId): Option[BlockResult] = {
+ def getLocalValues(blockId: BlockId): Option[BlockResult] =
+ getLocalValues(blockId, checkChecksumSeal = true)
+
+ /**
+ * `checkChecksumSeal` runs the read-side sealed-checksum self-check. Pass
`false` only from the
+ * producing store's own readback in `getOrElseUpdate` (not a consumer
read): in most cases the
+ * block is not sealed yet, and if it is a recompute of an already-sealed
block the divergent copy
+ * is still caught at master registration (a divergent report is rejected)
and at any later local
+ * read (which self-checks).
+ */
+ private def getLocalValues(blockId: BlockId, checkChecksumSeal: Boolean):
Option[BlockResult] = {
logDebug(s"Getting local block $blockId")
blockInfoManager.lockForReading(blockId) match {
case None =>
logDebug(s"Block $blockId was not found")
None
case Some(info) =>
+ // For a sealed block whose local copy does not match the sealed
checksum, skip the local
+ // copy and fall through to a remote (authoritative) location: it is a
stale replica the
+ // seal is evicting, but that eviction is async and may not have
landed. This self-check
+ // makes correctness independent of the eviction landing. Only
`verifySealed` blocks are
+ // checked; an unsealed block, or one checksummed only for
observation, is served as-is.
+ // This is only used with localCheckpoint, whose lineage is cut once
sealed, so a lost block
+ // is never recomputed: there is no later, differently-checksummed
re-materialization to
+ // reconcile here - a skipped local copy means the block is genuinely
gone, not superseded.
+ if (checkChecksumSeal && info.verifySealedChecksum &&
+ !localCopyMatchesSealedChecksum(blockId, info)) {
+ releaseLock(blockId, Option(TaskContext.get()))
+ return None
+ }
val level = info.level
logDebug(s"Level for block $blockId is $level")
val taskContext = Option(TaskContext.get())
@@ -1085,6 +1240,14 @@ private[spark] class BlockManager(
*
* Must be called while holding a read lock on the block.
* Releases the read lock upon exception; keeps the read lock upon
successful return.
+ *
+ * This is the serialized-bytes path, distinct from the value path
`getLocalValues`. Its callers
+ * are block replication (`doPutIterator`, `replicateBlock`) and serving a
block to a remote peer
+ * (`getLocalBlockData`); none is a local read of an RDD block's values. It
therefore does NOT run
+ * the read-side sealed-checksum self-check that `getLocalValues` does: for
a sealed RDD block,
+ * correctness is left to the master (`sealRddChecksums` /
`checksumSealRejectsUpdate`), and any
+ * other block reached here (broadcast, artifact, shuffle) is never sealed.
A read that must
+ * observe the seal must use `getLocalValues`.
*/
private def doGetLocalBytes(blockId: BlockId, info: BlockInfo): BlockData = {
val level = info.level
@@ -1411,9 +1574,11 @@ private[spark] class BlockManager(
blockId: RDDBlockId,
level: StorageLevel,
classTag: ClassTag[T],
- makeIterator: () => Iterator[T]): Either[BlockResult, Iterator[T]] = {
+ makeIterator: () => Iterator[T],
+ verifySealedChecksum: Boolean = false): Either[BlockResult, Iterator[T]]
= {
val isCacheVisible = isRDDBlockVisible(blockId)
- val res = getOrElseUpdate(blockId, level, classTag, makeIterator,
isCacheVisible)
+ val res = getOrElseUpdate(blockId, level, classTag, makeIterator,
isCacheVisible,
+ verifySealedChecksum = verifySealedChecksum)
if (res.isLeft && !isCacheVisible) {
// Block exists but not visible, report taskId -> blockId info to master.
master.updateRDDBlockTaskInfo(blockId, taskId)
@@ -1434,7 +1599,8 @@ private[spark] class BlockManager(
level: StorageLevel,
classTag: ClassTag[T],
makeIterator: () => Iterator[T],
- isCacheVisible: Boolean): Either[BlockResult, Iterator[T]] = {
+ isCacheVisible: Boolean,
+ verifySealedChecksum: Boolean = false): Either[BlockResult, Iterator[T]]
= {
// Track whether the data is computed or not, force to do the computation
later if need to.
// The reason we push the force computing later is that once the executor
is decommissioned we
// will have a better chance to replicate the cache block because of the
`checkShouldStore`
@@ -1459,7 +1625,8 @@ private[spark] class BlockManager(
// for same blockId could be different. And the reported accumulators
could be not matching
// the cached results.
// Initially we hold no locks on this block.
- doPutIterator(blockId, iterator, level, classTag, keepReadLock = true)
match {
+ doPutIterator(blockId, iterator, level, classTag, keepReadLock = true,
+ verifySealedChecksum = verifySealedChecksum) match {
case None =>
// doPut() didn't hand work back to us, so the block already existed
or was successfully
// stored. Therefore, we now hold a read lock on the block.
@@ -1467,7 +1634,11 @@ private[spark] class BlockManager(
// Force compute to report accumulator updates.
Utils.getIteratorSize(makeIterator())
}
- val blockResult = getLocalValues(blockId).getOrElse {
+ // checkChecksumSeal = false: this is the producing store's own
readback, not a consumer
+ // read, so the self-check would only add a master round-trip - in
most cases the block is
+ // not sealed yet here, and a recompute of an already-sealed block is
still caught at master
+ // registration (updateBlockInfo rejects the divergent report) and at
any later local read.
+ val blockResult = getLocalValues(blockId, checkChecksumSeal =
false).getOrElse {
// Since we held a read lock between the doPut() and get() calls,
the block should not
// have been evicted, so get() not returning the block indicates
some internal error.
releaseLock(blockId)
@@ -1554,10 +1725,14 @@ private[spark] class BlockManager(
blockId: BlockId,
bytes: ChunkedByteBuffer,
level: StorageLevel,
- tellMaster: Boolean = true): Boolean = {
+ tellMaster: Boolean = true,
+ sourceChecksum: Option[Long] = None,
+ sourceVerifySealedChecksum: Boolean = false): Boolean = {
require(bytes != null, "Bytes is null")
val blockStoreUpdater =
- ByteBufferBlockStoreUpdater(blockId, level, implicitly[ClassTag[T]],
bytes, tellMaster)
+ ByteBufferBlockStoreUpdater(blockId, level, implicitly[ClassTag[T]],
bytes, tellMaster,
+ sourceChecksum = sourceChecksum,
+ sourceVerifySealedChecksum = sourceVerifySealedChecksum)
blockStoreUpdater.save()
}
@@ -1679,10 +1854,12 @@ private[spark] class BlockManager(
level: StorageLevel,
classTag: ClassTag[T],
tellMaster: Boolean = true,
- keepReadLock: Boolean = false): Option[PartiallyUnrolledIterator[T]] = {
+ keepReadLock: Boolean = false,
+ verifySealedChecksum: Boolean = false):
Option[PartiallyUnrolledIterator[T]] = {
doPut(blockId, level, classTag, tellMaster = tellMaster, keepReadLock =
keepReadLock) { info =>
val startTimeNs = System.nanoTime()
var iteratorFromFailedMemoryStorePut:
Option[PartiallyUnrolledIterator[T]] = None
+ info.verifySealedChecksum = verifySealedChecksum
// Size of the block in bytes
var size = 0L
if (level.useMemory) {
@@ -1692,23 +1869,32 @@ private[spark] class BlockManager(
memoryStore.putIteratorAsValues(blockId, iterator(),
level.memoryMode, classTag) match {
case Right(s) =>
size = s
+ // A deserialized block resident in memory has no serialized
bytes, so no checksum is
+ // computed here; it gets one only if/when it is later
serialized - the spill-to-disk
+ // branch below, eviction (`dropFromMemory`), or on a replica
when received.
case Left(iter) =>
// Not enough space to unroll this block; drop to disk if
applicable
if (level.useDisk) {
logWarning(log"Persisting block ${MDC(BLOCK_ID, blockId)} to
disk instead.")
+ val checksumOpt = newRddBlockChecksum(blockId,
verifySealedChecksum)
diskStore.put(blockId) { channel =>
- val out = Channels.newOutputStream(channel)
+ val out = serializerManager.wrapForChecksum(
+ checksumOpt, Channels.newOutputStream(channel))
serializerManager.dataSerializeStream(blockId, out,
iter)(classTag)
}
+ recordChecksum(info, checksumOpt)
size = diskStore.getSize(blockId)
} else {
iteratorFromFailedMemoryStorePut = Some(iter)
}
}
} else { // !level.deserialized
- memoryStore.putIteratorAsBytes(blockId, iterator(), classTag,
level.memoryMode) match {
+ val checksumOpt = newRddBlockChecksum(blockId, verifySealedChecksum)
+ memoryStore.putIteratorAsBytes(
+ blockId, iterator(), classTag, level.memoryMode, checksumOpt)
match {
case Right(s) =>
size = s
+ recordChecksum(info, checksumOpt)
case Left(partiallySerializedValues) =>
// Not enough space to unroll this block; drop to disk if
applicable
if (level.useDisk) {
@@ -1717,6 +1903,9 @@ private[spark] class BlockManager(
val out = Channels.newOutputStream(channel)
partiallySerializedValues.finishWritingToStream(out)
}
+ // checksumOpt folds over the whole serialized stream
(in-memory portion plus the
+ // values finished to disk here), so its value now covers the
full block.
+ recordChecksum(info, checksumOpt)
size = diskStore.getSize(blockId)
} else {
iteratorFromFailedMemoryStorePut =
Some(partiallySerializedValues.valuesIterator)
@@ -1725,10 +1914,13 @@ private[spark] class BlockManager(
}
} else if (level.useDisk) {
+ val checksumOpt = newRddBlockChecksum(blockId, verifySealedChecksum)
diskStore.put(blockId) { channel =>
- val out = Channels.newOutputStream(channel)
+ val out = serializerManager.wrapForChecksum(
+ checksumOpt, Channels.newOutputStream(channel))
serializerManager.dataSerializeStream(blockId, out,
iterator())(classTag)
}
+ recordChecksum(info, checksumOpt)
size = diskStore.getSize(blockId)
}
@@ -1738,7 +1930,8 @@ private[spark] class BlockManager(
// Now that the block is in either the memory or disk store, tell the
master about it.
info.size = size
if (tellMaster && info.tellMaster) {
- reportBlockStatus(blockId, putBlockStatus)
+ // Carry the content checksum (set above for disk-serialized RDD
blocks).
+ reportBlockStatus(blockId, putBlockStatus, checksum = info.checksum)
}
addUpdatedBlockStatusToTaskMetrics(blockId, putBlockStatus)
logDebug(s"Put block $blockId locally took
${Utils.getUsedTimeNs(startTimeNs)}")
@@ -1746,7 +1939,9 @@ private[spark] class BlockManager(
val remoteStartTimeNs = System.nanoTime()
val bytesToReplicate = doGetLocalBytes(blockId, info)
try {
- replicate(blockId, bytesToReplicate, level, classTag)
+ replicate(blockId, bytesToReplicate, level, classTag,
+ sourceChecksum = info.checksum,
+ sourceVerifySealedChecksum = info.verifySealedChecksum)
} finally {
bytesToReplicate.dispose()
}
@@ -1895,7 +2090,9 @@ private[spark] class BlockManager(
getPeers(forceFetch = true)
try {
replicate(
- blockId, data, storageLevel, info.classTag, existingReplicas,
maxReplicationFailures)
+ blockId, data, storageLevel, info.classTag, existingReplicas,
maxReplicationFailures,
+ sourceChecksum = info.checksum,
+ sourceVerifySealedChecksum = info.verifySealedChecksum)
} finally {
logDebug(s"Releasing lock for $blockId")
releaseLockAndDispose(blockId, data)
@@ -1913,7 +2110,11 @@ private[spark] class BlockManager(
level: StorageLevel,
classTag: ClassTag[_],
existingReplicas: Set[BlockManagerId] = Set.empty,
- maxReplicationFailures: Option[Int] = None): Boolean = {
+ maxReplicationFailures: Option[Int] = None,
+ // Source block's content checksum + seal-path mark, forwarded to each
replica so it can be
+ // verified against the seal like a primary.
+ sourceChecksum: Option[Long] = None,
+ sourceVerifySealedChecksum: Boolean = false): Boolean = {
val maxReplicationFailureCount = maxReplicationFailures.getOrElse(
conf.get(config.STORAGE_MAX_REPLICATION_FAILURE))
@@ -1958,7 +2159,9 @@ private[spark] class BlockManager(
blockId,
buffer,
tLevel,
- classTag)
+ classTag,
+ sourceChecksum,
+ sourceVerifySealedChecksum)
logTrace(s"Replicated $blockId of ${data.size} bytes to $peer" +
s" in ${(System.nanoTime - onePeerStartTime).toDouble / 1e6} ms")
peersForReplication = peersForReplication.tail
@@ -2045,15 +2248,24 @@ private[spark] class BlockManager(
// Drop to disk, if storage level requires
if (level.useDisk && !diskStore.contains(blockId)) {
logInfo(log"Writing block ${MDC(BLOCK_ID, blockId)} to disk")
+ // Compute a content checksum for this serialize-to-disk on eviction
(same rule as the store
+ // path: only when requested). In practice the Left branch below is
reached only by a
+ // deserialized in-memory block, which is never on the seal path (the
mark is gated on a
+ // serialized level), so here the checksum is driven purely by the
global compute flag; a
+ // sealed (serialized) block dropped from memory takes the Right branch,
writing bytes whose
+ // checksum was already set at store time.
+ val checksumOpt = newRddBlockChecksum(blockId, info.verifySealedChecksum)
data() match {
case Left(elements) =>
diskStore.put(blockId) { channel =>
- val out = Channels.newOutputStream(channel)
+ val out = serializerManager.wrapForChecksum(
+ checksumOpt, Channels.newOutputStream(channel))
serializerManager.dataSerializeStream(
blockId,
out,
elements.iterator)(info.classTag.asInstanceOf[ClassTag[T]])
}
+ recordChecksum(info, checksumOpt)
case Right(bytes) =>
diskStore.putBytes(blockId, bytes)
}
@@ -2073,7 +2285,7 @@ private[spark] class BlockManager(
val status = getCurrentBlockStatus(blockId, info)
if (info.tellMaster) {
- reportBlockStatus(blockId, status, droppedMemorySize)
+ reportBlockStatus(blockId, status, droppedMemorySize, checksum =
info.checksum)
}
if (blockIsUpdated) {
addUpdatedBlockStatusToTaskMetrics(blockId, status)
diff --git
a/core/src/main/scala/org/apache/spark/storage/BlockManagerMaster.scala
b/core/src/main/scala/org/apache/spark/storage/BlockManagerMaster.scala
index 98bd52fc0886..1b27fc334d23 100644
--- a/core/src/main/scala/org/apache/spark/storage/BlockManagerMaster.scala
+++ b/core/src/main/scala/org/apache/spark/storage/BlockManagerMaster.scala
@@ -106,9 +106,10 @@ class BlockManagerMaster(
blockId: BlockId,
storageLevel: StorageLevel,
memSize: Long,
- diskSize: Long): Boolean = {
+ diskSize: Long,
+ checksum: Option[Long] = None): Boolean = {
val res = driverEndpoint.askSync[Boolean](
- UpdateBlockInfo(blockManagerId, blockId, storageLevel, memSize,
diskSize))
+ UpdateBlockInfo(blockManagerId, blockId, storageLevel, memSize,
diskSize, checksum))
logDebug(s"Updated info of block $blockId")
res
}
@@ -126,6 +127,26 @@ class BlockManagerMaster(
driverEndpoint.askSync[Boolean](GetRDDBlockVisibility(blockId))
}
+ /**
+ * The authoritative sealed checksum for an RDD block, or None if it is not
sealed. Returns None
+ * for a non-`RDDBlockId`: such blocks are never sealed.
+ */
+ def getSealedChecksum(blockId: BlockId): Option[Long] = {
+ blockId.asRDDId match {
+ case Some(rddBlockId) =>
driverEndpoint.askSync[Option[Long]](GetSealedChecksum(rddBlockId))
+ case None => None
+ }
+ }
+
+ /**
+ * Seal an RDD's per-block content checksums: for each materialized block
keep one version (the
+ * plurality checksum), evict divergent copies, and reject future divergent
registrations. Returns
+ * the count of present blocks that had no recorded checksum and so could
not be sealed.
+ */
+ def sealRddChecksums(rddId: Int): Int = {
+ driverEndpoint.askSync[Int](SealRddChecksums(rddId))
+ }
+
/** Get locations of the blockId from the driver */
def getLocations(blockId: BlockId): Seq[BlockManagerId] = {
driverEndpoint.askSync[Seq[BlockManagerId]](GetLocations(blockId))
diff --git
a/core/src/main/scala/org/apache/spark/storage/BlockManagerMasterEndpoint.scala
b/core/src/main/scala/org/apache/spark/storage/BlockManagerMasterEndpoint.scala
index 9d6539e09f45..58e1e899b11e 100644
---
a/core/src/main/scala/org/apache/spark/storage/BlockManagerMasterEndpoint.scala
+++
b/core/src/main/scala/org/apache/spark/storage/BlockManagerMasterEndpoint.scala
@@ -85,6 +85,17 @@ class BlockManagerMasterEndpoint(
// Mapping from block id to the set of block managers that have the block.
private val blockLocations = new JHashMap[BlockId,
mutable.HashSet[BlockManagerId]]
+ // Per-replica content checksum of RDD blocks, recorded alongside
`blockLocations` when RDD block
+ // checksums are enabled. Used to detect a block materialized inconsistently
by more than one
+ // task attempt (divergent bytes => different checksum across replicas).
Kept in lock-step with
+ // `blockLocations` on add/remove.
+ private val blockChecksums = new JHashMap[BlockId,
mutable.HashMap[BlockManagerId, Long]]
+
+ // The authoritative sealed checksum per finalized block. Once present, only
a copy with this
+ // checksum is admitted to `blockLocations`; divergent copies are rejected,
and reads self-check
+ // against it. Kept in lock-step with `blockLocations` on block/rdd/BM
removal.
+ private val sealedChecksums = new JHashMap[BlockId, Long]
+
// Mapping from task id to the set of rdd blocks which are generated from
the task.
private val tidToRddBlockIds = new mutable.HashMap[Long,
mutable.HashSet[RDDBlockId]]
// Record the RDD blocks which are not visible yet, a block will be removed
from this collection
@@ -141,7 +152,7 @@ class BlockManagerMasterEndpoint(
register(id, localDirs, maxOnHeapMemSize, maxOffHeapMemSize, endpoint,
isReRegister))
case _updateBlockInfo @
- UpdateBlockInfo(blockManagerId, blockId, storageLevel,
deserializedSize, size) =>
+ UpdateBlockInfo(blockManagerId, blockId, storageLevel,
deserializedSize, size, checksum) =>
@inline def handleResult(success: Boolean): Unit = {
// SPARK-30594: we should not post `SparkListenerBlockUpdated` when
updateBlockInfo
@@ -155,7 +166,8 @@ class BlockManagerMasterEndpoint(
if (blockId.isShuffle) {
updateShuffleBlockInfo(blockId, blockManagerId).foreach(handleResult)
} else {
- handleResult(updateBlockInfo(blockManagerId, blockId, storageLevel,
deserializedSize, size))
+ handleResult(
+ updateBlockInfo(blockManagerId, blockId, storageLevel,
deserializedSize, size, checksum))
}
case GetLocations(blockId) =>
@@ -241,6 +253,12 @@ class BlockManagerMasterEndpoint(
// Get the visibility status of a specific rdd block.
context.reply(isRDDBlockVisible(blockId))
+ case GetSealedChecksum(blockId) =>
+ context.reply(Option(sealedChecksums.get(blockId)).map(_.longValue))
+
+ case SealRddChecksums(rddId) =>
+ context.reply(sealRddChecksums(rddId))
+
case UpdateRDDBlockVisibility(taskId, visible) =>
// This is to report the information that whether rdd blocks computed by
task(with `taskId`)
// can be turned to be visible. This is reported by DAGScheduler right
after task completes.
@@ -360,6 +378,8 @@ class BlockManagerMasterEndpoint(
blocks.foreach { blockId =>
val bms: mutable.HashSet[BlockManagerId] = blockLocations.remove(blockId)
+ blockChecksums.remove(blockId)
+ sealedChecksums.remove(blockId)
if (trackingCacheVisibility) {
invisibleRDDBlocks.remove(blockId)
}
@@ -512,6 +532,7 @@ class BlockManagerMasterEndpoint(
val blockId = iterator.next
val locations = blockLocations.get(blockId)
locations -= blockManagerId
+ Option(blockChecksums.get(blockId)).foreach(_.remove(blockManagerId))
// De-register the block if none of the block managers have it.
Otherwise, if pro-active
// replication is enabled, and a block is either an RDD or a test block
(the latter is used
// for unit testing), we send a message to a randomly chosen executor
location to replicate
@@ -519,6 +540,8 @@ class BlockManagerMasterEndpoint(
// etc.) as replication doesn't make much sense in that context.
if (locations.isEmpty) {
blockLocations.remove(blockId)
+ blockChecksums.remove(blockId)
+ sealedChecksums.remove(blockId)
logWarning(log"No more replicas available for ${MDC(BLOCK_ID,
blockId)}!")
} else if (proactivelyReplicate && (blockId.isRDD ||
blockId.isInstanceOf[TestBlockId])) {
// As a heuristic, assume single executor failure to find out the
number of replicas that
@@ -797,7 +820,8 @@ class BlockManagerMasterEndpoint(
blockId: BlockId,
storageLevel: StorageLevel,
memSize: Long,
- diskSize: Long): Boolean = {
+ diskSize: Long,
+ checksum: Option[Long] = None): Boolean = {
logDebug(s"Updating block info on master ${blockId} for ${blockManagerId}")
if (!blockManagerInfo.contains(blockManagerId)) {
@@ -826,8 +850,30 @@ class BlockManagerMasterEndpoint(
}
if (storageLevel.isValid) {
+ // A sealed block admits only a copy matching its sealed checksum (see
+ // `checksumSealRejectsUpdate`). On reject, acknowledge the report but
drop the divergent copy
+ // and ask its executor to reclaim the local copy.
+ if (checksumSealRejectsUpdate(blockId, checksum)) {
+ evictReplica(blockId, blockManagerId)
+ // Return true (report accepted), not false: false means
"re-register", which would
+ // re-report this same block and hit this reject again, looping. The
copy is intentionally
+ // left out of the directory and reclaimed above; the read-side
self-check keeps reads
+ // correct meanwhile.
+ return true
+ }
val firstBlock = locations.isEmpty
locations.add(blockManagerId)
+ // Record this replica's content checksum (only present when one was
computed).
+ if (blockId.isRDD) {
+ checksum.foreach { c =>
+ var m = blockChecksums.get(blockId)
+ if (m == null) {
+ m = new mutable.HashMap[BlockManagerId, Long]
+ blockChecksums.put(blockId, m)
+ }
+ m.put(blockManagerId, c)
+ }
+ }
blockId.asRDDId.foreach { rddBlockId =>
(trackingCacheVisibility, firstBlock) match {
@@ -844,6 +890,7 @@ class BlockManagerMasterEndpoint(
}
} else {
locations.remove(blockManagerId)
+ Option(blockChecksums.get(blockId)).foreach(_.remove(blockManagerId))
}
if (blockId.isRDD && storageLevel.useDisk &&
externalShuffleServiceRddFetchEnabled) {
@@ -858,10 +905,74 @@ class BlockManagerMasterEndpoint(
// Remove the block from master tracking if it has been removed on all
endpoints.
if (locations.isEmpty) {
blockLocations.remove(blockId)
+ blockChecksums.remove(blockId)
+ sealedChecksums.remove(blockId)
}
true
}
+ /**
+ * Whether a report of `blockId` carrying content checksum `checksum` must
be kept out of the
+ * directory: the block is sealed and this copy does not match the sealed
value. A `None` report
+ * of a sealed block also fails (a sealable block always carries a checksum,
so `None` is
+ * anomalous and must not enter the directory unverified).
+ */
+ private def checksumSealRejectsUpdate(blockId: BlockId, checksum:
Option[Long]): Boolean =
+ sealedChecksums.containsKey(blockId) &&
+ !checksum.contains(sealedChecksums.get(blockId).longValue)
+
+ /**
+ * Ask an executor to drop its local copy of a block, fire-and-forget: the
reply is discarded
+ * because correctness comes from the directory (this block manager is
already removed from
+ * `blockLocations`) plus the read-side self-check, not from the eviction
landing. Used by the
+ * seal to reclaim divergent replicas (`sealRddChecksums` losers and the
reject path).
+ */
+ private def evictReplica(blockId: BlockId, bmId: BlockManagerId): Unit = {
+ blockManagerInfo.get(bmId).foreach { bm =>
+ bm.storageEndpoint.ask[Boolean](RemoveBlock(blockId))
+ ()
+ }
+ }
+
+ /**
+ * Seal an RDD's per-block content checksums: for each of its blocks with
recorded per-replica
+ * checksums, pick one checksum value as authoritative, evict the replicas
that disagree with it
+ * (drop them from the directory and ask their executors to remove the local
copy), and record the
+ * authoritative value so later divergent registrations are rejected and
reads can self-check.
+ *
+ * Runs on the message-handler thread, so it is atomic w.r.t. other
`updateBlockInfo` for these
+ * blocks. The per-executor removals are fire-and-forget (best-effort space
reclamation); reads
+ * of an orphan that has not been removed yet are caught by the read-side
self-check.
+ *
+ * Returns the number of present blocks that had no recorded checksum and so
could not be sealed.
+ */
+ private def sealRddChecksums(rddId: Int): Int = {
+ val rddBlocks =
blockLocations.asScala.keys.flatMap(_.asRDDId).filter(_.rddId == rddId).toSeq
+ var unchecksummed = 0
+ rddBlocks.foreach { blockId =>
+ val perReplica = blockChecksums.get(blockId)
+ if (perReplica != null && perReplica.nonEmpty) {
+ // Any surviving version is a valid snapshot, so correctness does not
require a particular
+ // winner. We keep the plurality checksum (ties broken arbitrarily) so
the survivor is the
+ // best-replicated version - the one most resilient to replica loss,
which matters here
+ // because a lost local-checkpoint block cannot be recomputed.
+ val winner = perReplica.values.groupBy(identity).maxBy(_._2.size)._1
+ sealedChecksums.put(blockId, winner)
+ val losers = perReplica.collect { case (bmId, c) if c != winner =>
bmId }.toSeq
+ losers.foreach { bmId =>
+ perReplica.remove(bmId)
+ Option(blockLocations.get(blockId)).foreach(_.remove(bmId))
+ evictReplica(blockId, bmId)
+ }
+ } else {
+ // Present but with no recorded content checksum, so it cannot be
sealed. The caller
+ // decides what to make of it (e.g. a coverage warning).
+ unchecksummed += 1
+ }
+ }
+ unchecksummed
+ }
+
private def getLocations(blockId: BlockId): Seq[BlockManagerId] = {
if (blockLocations.containsKey(blockId)) blockLocations.get(blockId).toSeq
else Seq.empty
}
diff --git
a/core/src/main/scala/org/apache/spark/storage/BlockManagerMessages.scala
b/core/src/main/scala/org/apache/spark/storage/BlockManagerMessages.scala
index 7fb145556a11..d32895700f43 100644
--- a/core/src/main/scala/org/apache/spark/storage/BlockManagerMessages.scala
+++ b/core/src/main/scala/org/apache/spark/storage/BlockManagerMessages.scala
@@ -80,7 +80,10 @@ private[spark] object BlockManagerMessages {
var blockId: BlockId,
var storageLevel: StorageLevel,
var memSize: Long,
- var diskSize: Long)
+ var diskSize: Long,
+ // Per-replica content checksum of the serialized (pre-encryption) block
bytes, or None when
+ // RDD block checksums are disabled.
+ var checksum: Option[Long] = None)
extends ToBlockManagerMaster
with Externalizable {
@@ -92,6 +95,8 @@ private[spark] object BlockManagerMessages {
storageLevel.writeExternal(out)
out.writeLong(memSize)
out.writeLong(diskSize)
+ out.writeBoolean(checksum.isDefined)
+ checksum.foreach(out.writeLong)
}
override def readExternal(in: ObjectInput): Unit = Utils.tryOrIOException {
@@ -100,6 +105,7 @@ private[spark] object BlockManagerMessages {
storageLevel = StorageLevel(in)
memSize = in.readLong()
diskSize = in.readLong()
+ checksum = if (in.readBoolean()) Some(in.readLong()) else None
}
}
@@ -109,6 +115,13 @@ private[spark] object BlockManagerMessages {
case class GetRDDBlockVisibility(blockId: RDDBlockId) extends
ToBlockManagerMaster
+ // The authoritative sealed checksum for an RDD block, or None if it is not
sealed.
+ case class GetSealedChecksum(blockId: RDDBlockId) extends
ToBlockManagerMaster
+
+ // Seal an RDD's per-block content checksums: per block keep the plurality
checksum, evict
+ // divergent copies, and reject future divergent registrations.
+ case class SealRddChecksums(rddId: Int) extends ToBlockManagerMaster
+
case class GetLocations(blockId: BlockId) extends ToBlockManagerMaster
case class GetLocationsAndStatus(blockId: BlockId, requesterHost: String)
diff --git
a/core/src/main/scala/org/apache/spark/storage/memory/MemoryStore.scala
b/core/src/main/scala/org/apache/spark/storage/memory/MemoryStore.scala
index 3981006c7ffe..262eeab5fcf8 100644
--- a/core/src/main/scala/org/apache/spark/storage/memory/MemoryStore.scala
+++ b/core/src/main/scala/org/apache/spark/storage/memory/MemoryStore.scala
@@ -20,6 +20,7 @@ package org.apache.spark.storage.memory
import java.io.OutputStream
import java.nio.ByteBuffer
import java.util.LinkedHashMap
+import java.util.zip.Checksum
import scala.collection.mutable
import scala.collection.mutable.ArrayBuffer
@@ -346,7 +347,8 @@ private[spark] class MemoryStore(
blockId: BlockId,
values: Iterator[T],
classTag: ClassTag[T],
- memoryMode: MemoryMode): Either[PartiallySerializedBlock[T], Long] = {
+ memoryMode: MemoryMode,
+ checksum: Option[Checksum] = None): Either[PartiallySerializedBlock[T],
Long] = {
require(!contains(blockId), s"Block $blockId is already present in the
MemoryStore")
@@ -363,7 +365,7 @@ private[spark] class MemoryStore(
}
val valuesHolder = new SerializedValuesHolder[T](blockId, chunkSize,
classTag,
- memoryMode, serializerManager)
+ memoryMode, serializerManager, checksum)
val res = putIterator(blockId, values, classTag, memoryMode, valuesHolder)
match {
case Right(storedSize) => Right(storedSize)
@@ -730,7 +732,8 @@ private class SerializedValuesHolder[T](
chunkSize: Int,
classTag: ClassTag[T],
memoryMode: MemoryMode,
- serializerManager: SerializerManager) extends ValuesHolder[T] {
+ serializerManager: SerializerManager,
+ checksum: Option[Checksum] = None) extends ValuesHolder[T] {
val allocator = memoryMode match {
case MemoryMode.ON_HEAP => ByteBuffer.allocate _
case MemoryMode.OFF_HEAP => Platform.allocateDirectBuffer _
@@ -742,7 +745,10 @@ private class SerializedValuesHolder[T](
val serializationStream: SerializationStream = {
val autoPick = !blockId.isInstanceOf[StreamBlockId]
val ser = serializerManager.getSerializer(classTag, autoPick).newInstance()
- ser.serializeStream(serializerManager.wrapForCompression(blockId,
redirectableStream))
+ // Optionally fold a content checksum over the serialized+compressed
plaintext (composed above
+ // the sink, below compression).
+ val sink = serializerManager.wrapForChecksum(checksum, redirectableStream)
+ ser.serializeStream(serializerManager.wrapForCompression(blockId, sink))
}
override def storeValue(value: T): Unit = {
diff --git
a/core/src/test/scala/org/apache/spark/network/BlockTransferServiceSuite.scala
b/core/src/test/scala/org/apache/spark/network/BlockTransferServiceSuite.scala
index f9a1b778b4ea..ca5924617798 100644
---
a/core/src/test/scala/org/apache/spark/network/BlockTransferServiceSuite.scala
+++
b/core/src/test/scala/org/apache/spark/network/BlockTransferServiceSuite.scala
@@ -89,7 +89,9 @@ class BlockTransferServiceSuite extends SparkFunSuite with
TimeLimits {
blockId: BlockId,
blockData: ManagedBuffer,
level: StorageLevel,
- classTag: ClassTag[_]): Future[Unit] = {
+ classTag: ClassTag[_],
+ checksum: Option[Long] = None,
+ verifySealedChecksum: Boolean = false): Future[Unit] = {
// This method is unused in this test
throw new UnsupportedOperationException("uploadBlock")
}
diff --git
a/core/src/test/scala/org/apache/spark/rdd/LocalCheckpointSuite.scala
b/core/src/test/scala/org/apache/spark/rdd/LocalCheckpointSuite.scala
index 9768da09a9f8..030c88d02dec 100644
--- a/core/src/test/scala/org/apache/spark/rdd/LocalCheckpointSuite.scala
+++ b/core/src/test/scala/org/apache/spark/rdd/LocalCheckpointSuite.scala
@@ -21,7 +21,8 @@ import scala.concurrent.duration._
import org.scalatest.concurrent.Eventually.{eventually, interval, timeout}
-import org.apache.spark.{LocalSparkContext, SparkContext, SparkException,
SparkFunSuite}
+import org.apache.spark.{LocalSparkContext, SparkConf, SparkContext,
SparkException, SparkFunSuite}
+import
org.apache.spark.internal.config.{LOCAL_CHECKPOINT_VERIFY_CHECKSUM_ENABLED,
LOCAL_CHECKPOINT_VERIFY_CHECKSUM_FORCE_SERIALIZED}
import org.apache.spark.storage.{RDDBlockId, StorageLevel}
/**
@@ -36,7 +37,7 @@ class LocalCheckpointSuite extends SparkFunSuite with
LocalSparkContext {
}
test("transform storage level") {
- val transform = LocalRDDCheckpointData.transformStorageLevel _
+ val transform = (level: StorageLevel) =>
LocalRDDCheckpointData.transformStorageLevel(level)
assert(transform(StorageLevel.NONE) === StorageLevel.DISK_ONLY)
assert(transform(StorageLevel.MEMORY_ONLY) ===
StorageLevel.MEMORY_AND_DISK)
assert(transform(StorageLevel.MEMORY_ONLY_SER) ===
StorageLevel.MEMORY_AND_DISK_SER)
@@ -51,6 +52,38 @@ class LocalCheckpointSuite extends SparkFunSuite with
LocalSparkContext {
assert(transform(StorageLevel.MEMORY_AND_DISK_SER_2) ===
StorageLevel.MEMORY_AND_DISK_SER_2)
}
+ test("checksum verification is marked only for serialized checkpoints") {
+ // Verification off: never marked, default (deserialized) level untouched.
+ assert(markAndLevel() === (false,
LocalRDDCheckpointData.DEFAULT_STORAGE_LEVEL))
+
+ // Verification on, default deserialized level: not marked (nothing to
checksum), level kept.
+ assert(
+ markAndLevel(LOCAL_CHECKPOINT_VERIFY_CHECKSUM_ENABLED.key -> "true") ===
+ (false, LocalRDDCheckpointData.DEFAULT_STORAGE_LEVEL))
+ }
+
+ test("checksum verification force-serialized bumps a default checkpoint to a
serialized level") {
+ // Verification on plus force-serialized: default level bumped to
serialized and marked.
+ val (forcedMark, forcedLevel) = markAndLevel(
+ LOCAL_CHECKPOINT_VERIFY_CHECKSUM_ENABLED.key -> "true",
+ LOCAL_CHECKPOINT_VERIFY_CHECKSUM_FORCE_SERIALIZED.key -> "true")
+ assert(forcedMark)
+ assert(!forcedLevel.deserialized)
+ assert(forcedLevel === StorageLevel.MEMORY_AND_DISK_SER)
+ }
+
+ test("checksum verification is marked for an explicit serialized
checkpoint") {
+ resetSparkContext()
+ val conf = new SparkConf().setMaster("local[2]").setAppName("test")
+ .set(LOCAL_CHECKPOINT_VERIFY_CHECKSUM_ENABLED.key, "true")
+ sc = new SparkContext(conf)
+
+ // An explicit serialized level (e.g. DISK_ONLY) is verifiable without the
force flag.
+ val rdd = newRdd.persist(StorageLevel.DISK_ONLY).localCheckpoint()
+ assert(rdd.verifyCheckpointChecksums)
+ assert(!rdd.getStorageLevel.deserialized)
+ }
+
test("basic lineage truncation") {
val numPartitions = 4
val parallelRdd = sc.parallelize(1 to 100, numPartitions)
@@ -188,6 +221,19 @@ class LocalCheckpointSuite extends SparkFunSuite with
LocalSparkContext {
}
}
+ /**
+ * Rebuild the context with the given configs, then localCheckpoint a fresh
RDD and report
+ * whether it was marked for verification and what storage level it ended up
with.
+ */
+ private def markAndLevel(configs: (String, String)*): (Boolean,
StorageLevel) = {
+ resetSparkContext()
+ val conf = new SparkConf().setMaster("local[2]").setAppName("test")
+ configs.foreach { case (k, v) => conf.set(k, v) }
+ sc = new SparkContext(conf)
+ val rdd = newRdd.localCheckpoint()
+ (rdd.verifyCheckpointChecksums, rdd.getStorageLevel)
+ }
+
/**
* Helper method to create a simple RDD.
*/
diff --git
a/core/src/test/scala/org/apache/spark/serializer/SerializerManagerSuite.scala
b/core/src/test/scala/org/apache/spark/serializer/SerializerManagerSuite.scala
new file mode 100644
index 000000000000..597902b121f1
--- /dev/null
+++
b/core/src/test/scala/org/apache/spark/serializer/SerializerManagerSuite.scala
@@ -0,0 +1,46 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.serializer
+
+import java.io.ByteArrayOutputStream
+
+import org.apache.spark.{SparkConf, SparkFunSuite}
+import org.apache.spark.network.shuffle.checksum.ShuffleChecksumHelper
+
+class SerializerManagerSuite extends SparkFunSuite {
+
+ test("wrapForChecksum folds a deterministic content checksum over the
written bytes") {
+ val sparkConf = new SparkConf(false)
+ val serManager = new SerializerManager(new JavaSerializer(sparkConf),
sparkConf)
+ def checksumOf(bytes: Array[Byte]): Long = {
+ val checksum = ShuffleChecksumHelper.getChecksumByAlgorithm("CRC32C")
+ val out = serManager.wrapForChecksum(checksum, new
ByteArrayOutputStream())
+ out.write(bytes)
+ out.close()
+ checksum.getValue
+ }
+ val a = Array[Byte](1, 2, 3, 4, 5)
+ val b = Array[Byte](1, 2, 3, 4, 5)
+ val c = Array[Byte](1, 2, 3, 4, 6)
+ // Identical content gives an identical checksum (so equal replicas
survive a seal); divergent
+ // content gives a different one (so a divergently-materialized copy is
detectable).
+ assert(checksumOf(a) === checksumOf(b))
+ assert(checksumOf(a) !== checksumOf(c))
+ assert(checksumOf(Array.empty[Byte]) === checksumOf(Array.empty[Byte]))
+ }
+}
diff --git
a/core/src/test/scala/org/apache/spark/storage/BlockManagerDecommissionUnitSuite.scala
b/core/src/test/scala/org/apache/spark/storage/BlockManagerDecommissionUnitSuite.scala
index ab6c19575d23..4d60ea0dca61 100644
---
a/core/src/test/scala/org/apache/spark/storage/BlockManagerDecommissionUnitSuite.scala
+++
b/core/src/test/scala/org/apache/spark/storage/BlockManagerDecommissionUnitSuite.scala
@@ -217,12 +217,14 @@ class BlockManagerDecommissionUnitSuite extends
SparkFunSuite with Matchers {
// Simulate FileNotFoundException wrap inside SparkException
when(
blockTransferService
- .uploadBlock(mc.any(), mc.any(), mc.any(), mc.any(), mc.any(),
mc.any(), mc.isNull()))
+ .uploadBlock(mc.any(), mc.any(), mc.any(), mc.any(), mc.any(),
mc.any(), mc.isNull(),
+ mc.any(), mc.any()))
.thenReturn(Future.failed(
new java.io.IOException("boop", new FileNotFoundException("file not
found"))))
when(
blockTransferService
- .uploadBlockSync(mc.any(), mc.any(), mc.any(), mc.any(), mc.any(),
mc.any(), mc.isNull()))
+ .uploadBlockSync(mc.any(), mc.any(), mc.any(), mc.any(), mc.any(),
mc.any(), mc.isNull(),
+ mc.any(), mc.any()))
.thenCallRealMethod()
when(bm.blockTransferService).thenReturn(blockTransferService)
@@ -253,12 +255,13 @@ class BlockManagerDecommissionUnitSuite extends
SparkFunSuite with Matchers {
val blockTransferService = mock(classOf[BlockTransferService])
// Simulate BlockSavedOnDecommissionedBlockManagerException
when(blockTransferService.uploadBlock(
- mc.any(), mc.any(), mc.eq(exe1.executorId), mc.any(), mc.any(),
mc.any(), mc.isNull()))
+ mc.any(), mc.any(), mc.eq(exe1.executorId), mc.any(), mc.any(),
mc.any(), mc.isNull(),
+ mc.any(), mc.any()))
.thenReturn(
Future.failed(new
RuntimeException("BlockSavedOnDecommissionedBlockManagerException"))
)
when(blockTransferService.uploadBlockSync(
- mc.any(), mc.any(), mc.any(), mc.any(), mc.any(), mc.any(), mc.isNull()))
+ mc.any(), mc.any(), mc.any(), mc.any(), mc.any(), mc.any(), mc.isNull(),
mc.any(), mc.any()))
.thenCallRealMethod()
when(bm.blockTransferService).thenReturn(blockTransferService)
@@ -268,10 +271,10 @@ class BlockManagerDecommissionUnitSuite extends
SparkFunSuite with Matchers {
validateDecommissionTimestampsOnManager(bmDecomManager)
verify(blockTransferService, times(1))
.uploadBlock(mc.any(), mc.any(), mc.eq(exe1.executorId),
- mc.any(), mc.any(), mc.any(), mc.isNull())
+ mc.any(), mc.any(), mc.any(), mc.isNull(), mc.any(), mc.any())
verify(blockTransferService, times(1))
.uploadBlock(mc.any(), mc.any(), mc.eq(exe2.executorId),
- mc.any(), mc.any(), mc.any(), mc.isNull())
+ mc.any(), mc.any(), mc.any(), mc.isNull(), mc.any(), mc.any())
}
test("SPARK-54796: block decom manager handles
ShuffleManagerNotInitializedException " +
@@ -295,7 +298,7 @@ class BlockManagerDecommissionUnitSuite extends
SparkFunSuite with Matchers {
// Simulate ShuffleManagerNotInitializedException on first attempt,
// then succeed on retry to the same peer (transient condition resolved)
when(blockTransferService.uploadBlock(
- mc.any(), mc.any(), mc.any(), mc.any(), mc.any(), mc.any(), mc.isNull()))
+ mc.any(), mc.any(), mc.any(), mc.any(), mc.any(), mc.any(), mc.isNull(),
mc.any(), mc.any()))
.thenAnswer(new Answer[Future[Unit]] {
override def answer(invocation: InvocationOnMock): Future[Unit] = {
val attempt = uploadAttempts.incrementAndGet()
@@ -309,7 +312,7 @@ class BlockManagerDecommissionUnitSuite extends
SparkFunSuite with Matchers {
}
})
when(blockTransferService.uploadBlockSync(
- mc.any(), mc.any(), mc.any(), mc.any(), mc.any(), mc.any(), mc.isNull()))
+ mc.any(), mc.any(), mc.any(), mc.any(), mc.any(), mc.any(), mc.isNull(),
mc.any(), mc.any()))
.thenCallRealMethod()
when(bm.blockTransferService).thenReturn(blockTransferService)
@@ -321,7 +324,7 @@ class BlockManagerDecommissionUnitSuite extends
SparkFunSuite with Matchers {
// all to the same peer since the thread keeps running
verify(blockTransferService, times(3))
.uploadBlock(mc.any(), mc.any(), mc.eq(exe1.executorId),
- mc.any(), mc.any(), mc.any(), mc.isNull())
+ mc.any(), mc.any(), mc.any(), mc.isNull(), mc.any(), mc.any())
}
test("block decom manager handles IO failures") {
@@ -338,7 +341,8 @@ class BlockManagerDecommissionUnitSuite extends
SparkFunSuite with Matchers {
val blockTransferService = mock(classOf[BlockTransferService])
// Simulate an ambiguous IO error (e.g. block could be gone, connection
failed, etc.)
when(blockTransferService.uploadBlockSync(
- mc.any(), mc.any(), mc.any(), mc.any(), mc.any(), mc.any(),
mc.isNull())).thenThrow(
+ mc.any(), mc.any(), mc.any(), mc.any(), mc.any(), mc.any(), mc.isNull(),
+ mc.any(), mc.any())).thenThrow(
new java.io.IOException("boop")
)
@@ -372,7 +376,8 @@ class BlockManagerDecommissionUnitSuite extends
SparkFunSuite with Matchers {
val blockTransferService = mock(classOf[BlockTransferService])
// Simulate an ambiguous IO error (e.g. block could be gone, connection
failed, etc.)
when(blockTransferService.uploadBlockSync(
- mc.any(), mc.any(), mc.any(), mc.any(), mc.any(), mc.any(),
mc.isNull())).thenThrow(
+ mc.any(), mc.any(), mc.any(), mc.any(), mc.any(), mc.any(), mc.isNull(),
+ mc.any(), mc.any())).thenThrow(
new java.io.IOException("boop")
)
@@ -418,7 +423,7 @@ class BlockManagerDecommissionUnitSuite extends
SparkFunSuite with Matchers {
mc.eq(storedBlockId1), mc.any(), mc.any(), mc.eq(Some(3)))
verify(blockTransferService, times(2))
.uploadBlockSync(mc.eq("host2"), mc.eq(bmPort), mc.eq("exec2"),
mc.any(), mc.any(),
- mc.eq(StorageLevel.DISK_ONLY), mc.isNull())
+ mc.eq(StorageLevel.DISK_ONLY), mc.isNull(), mc.any(), mc.any())
// Since we never "finish" the RDD blocks, make sure the time is
always moving forward.
assert(bmDecomManager.rddBlocksLeft)
previousRDDTime match {
diff --git
a/core/src/test/scala/org/apache/spark/storage/BlockManagerSuite.scala
b/core/src/test/scala/org/apache/spark/storage/BlockManagerSuite.scala
index 933fd0b8c0e7..aadf52d81853 100644
--- a/core/src/test/scala/org/apache/spark/storage/BlockManagerSuite.scala
+++ b/core/src/test/scala/org/apache/spark/storage/BlockManagerSuite.scala
@@ -17,7 +17,7 @@
package org.apache.spark.storage
-import java.io.{File, InputStream, IOException}
+import java.io.{ByteArrayInputStream, ByteArrayOutputStream, File,
InputStream, IOException, ObjectInputStream, ObjectOutputStream}
import java.nio.ByteBuffer
import java.nio.file.Files
import java.util.UUID
@@ -52,7 +52,7 @@ import org.apache.spark.memory.{MemoryMode,
UnifiedMemoryManager}
import org.apache.spark.network.{BlockDataManager, BlockTransferService,
TransportContext}
import org.apache.spark.network.buffer.{FileSegmentManagedBuffer,
ManagedBuffer, NioManagedBuffer}
import org.apache.spark.network.client.{RpcResponseCallback, TransportClient}
-import org.apache.spark.network.netty.{NettyBlockTransferService,
SparkTransportConf}
+import org.apache.spark.network.netty.{BlockReplicationMetadata,
NettyBlockTransferService, SparkTransportConf}
import org.apache.spark.network.server.{NoOpRpcHandler, TransportServer,
TransportServerBootstrap}
import org.apache.spark.network.shuffle.{BlockFetchingListener,
DownloadFileManager, ExecutorDiskUtils, ExternalBlockStoreClient}
import org.apache.spark.network.shuffle.protocol.{BlockTransferMessage,
RegisterExecutor}
@@ -334,7 +334,7 @@ class BlockManagerSuite extends SparkFunSuite with Matchers
with PrivateMethodTe
eventually(timeout(5.seconds)) {
// For non-shuffle blocks, it should just report block manager id.
verify(master, times(1))
- .updateBlockInfo(mc.eq(blockManagerId), mc.any(), mc.any(),
mc.any(), mc.any())
+ .updateBlockInfo(mc.eq(blockManagerId), mc.any(), mc.any(),
mc.any(), mc.any(), mc.any())
}
bm.reportBlockStatus(BlockId("shuffle_0_0_0.index"), BlockStatus.empty)
bm.reportBlockStatus(BlockId("shuffle_0_0_0.data"), BlockStatus.empty)
@@ -347,7 +347,7 @@ class BlockManagerSuite extends SparkFunSuite with Matchers
with PrivateMethodTe
(blockManagerId, 3)
}
verify(master, times(expectedTimes))
- .updateBlockInfo(mc.eq(expectedBMId), mc.any(), mc.any(), mc.any(),
mc.any())
+ .updateBlockInfo(mc.eq(expectedBMId), mc.any(), mc.any(), mc.any(),
mc.any(), mc.any())
}
}
}
@@ -741,7 +741,7 @@ class BlockManagerSuite extends SparkFunSuite with Matchers
with PrivateMethodTe
val storageLevelCaptor =
ArgumentCaptor.forClass(classOf[StorageLevel]).asInstanceOf[ArgumentCaptor[StorageLevel]]
verify(master, atLeastOnce()).updateBlockInfo(mc.eq(store.blockManagerId),
mc.eq(blockId),
- storageLevelCaptor.capture(), memSizeCaptor.capture(),
diskSizeCaptor.capture())
+ storageLevelCaptor.capture(), memSizeCaptor.capture(),
diskSizeCaptor.capture(), mc.any())
assertSizeReported(memSizeCaptor, removedFromMemory)
assertSizeReported(diskSizeCaptor, removedFromDisk)
assert(storageLevelCaptor.getValue.replication == 0)
@@ -749,7 +749,7 @@ class BlockManagerSuite extends SparkFunSuite with Matchers
with PrivateMethodTe
private def assertUpdateBlockInfoNotReported(store: BlockManager, blockId:
BlockId): Unit = {
verify(master, never()).updateBlockInfo(mc.eq(store.blockManagerId),
mc.eq(blockId),
- mc.any[StorageLevel](), mc.anyInt(), mc.anyInt())
+ mc.any[StorageLevel](), mc.anyInt(), mc.anyInt(), mc.any())
}
test("reregistration on heart beat") {
@@ -1437,6 +1437,35 @@ class BlockManagerSuite extends SparkFunSuite with
Matchers with PrivateMethodTe
}
}
+ test("verifyOnReplication recomputes over the received bytes on the
stream-upload path") {
+ // The stream-upload path (putBlockDataAsStream ->
TempFileBasedBlockStoreUpdater) moves the
+ // temp file into the block store, so the verifyOnReplication recompute
must read the received
+ // bytes before that move, not after (else FileNotFoundException). Drive
that path directly with
+ // a seal-marked RDD block and the flag on.
+ val verifyConf =
+ new
SparkConf(false).set(STORAGE_RDD_BLOCK_CHECKSUM_VERIFY_ON_REPLICATION, true)
+ val store = makeBlockManager(20000, "exec1", testConf = Some(verifyConf))
+ try {
+ val message = "message"
+ val ser = serializer.newInstance().serialize(message).array()
+ val blockId = RDDBlockId(40, 0)
+ // A source checksum + seal mark, as a replicating peer would send.
+ val callback = store.putBlockDataAsStream(
+ blockId, StorageLevel.DISK_ONLY, ClassTag(message.getClass),
+ checksum = Some(123L), verifySealedChecksum = true)
+ callback.onData("0", ByteBuffer.wrap(ser))
+ // Pre-fix this threw FileNotFoundException from the recompute reading
the moved temp file.
+ callback.onComplete("0")
+ assert(store.getStatus(blockId).exists(_.diskSize > 0))
+ val info = store.blockInfoManager.get(blockId).get
+ // The recomputed checksum (over the received bytes) was recorded and
the mark propagated.
+ assert(info.checksum.isDefined)
+ assert(info.verifySealedChecksum)
+ } finally {
+ store.stop()
+ }
+ }
+
test("turn off updated block statuses") {
val conf = new SparkConf()
conf.set(TASK_METRICS_TRACK_UPDATED_BLOCK_STATUSES, false)
@@ -1821,7 +1850,9 @@ class BlockManagerSuite extends SparkFunSuite with
Matchers with PrivateMethodTe
blockId: BlockId,
blockData: ManagedBuffer,
level: StorageLevel,
- classTag: ClassTag[_]): Future[Unit] = {
+ classTag: ClassTag[_],
+ checksum: Option[Long] = None,
+ verifySealedChecksum: Boolean = false): Future[Unit] = {
throw new InterruptedException("Intentional interrupt")
}
}
@@ -2418,6 +2449,255 @@ class BlockManagerSuite extends SparkFunSuite with
Matchers with PrivateMethodTe
verify(master, times(2)).updateRDDBlockTaskInfo(blockId, 1)
}
+ // Local-checkpoint content-checksum verification (dedup-and-seal). The
master keeps one
+ // authoritative checksum per RDD block, evicts divergent replicas, rejects
later divergent
+ // registrations, and reports partitions it cannot check. See
SerializerManager.wrapForChecksum,
+ // BlockManagerMasterEndpoint.sealRddChecksums, and LocalRDDCheckpointData.
+ test("sealRddChecksums keeps the plurality checksum and evicts divergent
replicas") {
+ val store1 = makeBlockManager(20000, "exec1")
+ val store2 = makeBlockManager(20000, "exec2")
+ val store3 = makeBlockManager(20000, "exec3")
+ val blockId = RDDBlockId(7, 0)
+ // Two replicas agree on checksum 100, one diverged at 200.
+ master.updateBlockInfo(
+ store1.blockManagerId, blockId, StorageLevel.DISK_ONLY, 0, 100,
Some(100L))
+ master.updateBlockInfo(
+ store2.blockManagerId, blockId, StorageLevel.DISK_ONLY, 0, 100,
Some(100L))
+ master.updateBlockInfo(
+ store3.blockManagerId, blockId, StorageLevel.DISK_ONLY, 0, 100,
Some(200L))
+ assert(master.getLocations(blockId).toSet ===
+ Set(store1.blockManagerId, store2.blockManagerId, store3.blockManagerId))
+
+ assert(master.sealRddChecksums(7) === 0)
+ assert(master.getSealedChecksum(blockId) === Some(100L))
+ assert(master.getLocations(blockId).toSet ===
+ Set(store1.blockManagerId, store2.blockManagerId))
+ }
+
+ test("sealRddChecksums rejects later divergent registrations and admits
matching ones") {
+ val store1 = makeBlockManager(20000, "exec1")
+ val store2 = makeBlockManager(20000, "exec2")
+ val blockId = RDDBlockId(8, 0)
+ master.updateBlockInfo(
+ store1.blockManagerId, blockId, StorageLevel.DISK_ONLY, 0, 100,
Some(42L))
+ assert(master.sealRddChecksums(8) === 0)
+ assert(master.getSealedChecksum(blockId) === Some(42L))
+
+ // A divergent checksum is acknowledged but not admitted to the directory.
+ master.updateBlockInfo(
+ store2.blockManagerId, blockId, StorageLevel.DISK_ONLY, 0, 100,
Some(99L))
+ assert(!master.getLocations(blockId).contains(store2.blockManagerId))
+ // A checksum-less report of a sealed block is also rejected (a sealed
block must always
+ // report its checksum; a None here is anomalous and must not enter the
directory unverified).
+ master.updateBlockInfo(
+ store2.blockManagerId, blockId, StorageLevel.DISK_ONLY, 0, 100, None)
+ assert(!master.getLocations(blockId).contains(store2.blockManagerId))
+ // A matching checksum is admitted.
+ master.updateBlockInfo(
+ store2.blockManagerId, blockId, StorageLevel.DISK_ONLY, 0, 100,
Some(42L))
+ assert(master.getLocations(blockId).contains(store2.blockManagerId))
+ }
+
+ test("sealRddChecksums reports materialized partitions that carry no
checksum") {
+ val store = makeBlockManager(20000, "exec1")
+ // Present in the directory but with no recorded checksum (a deserialized
in-memory block, or a
+ // block materialized before localCheckpoint marked the RDD).
+ master.updateBlockInfo(
+ store.blockManagerId, RDDBlockId(9, 0), StorageLevel.DISK_ONLY, 0, 100,
None)
+ master.updateBlockInfo(
+ store.blockManagerId, RDDBlockId(9, 1), StorageLevel.DISK_ONLY, 0, 100,
Some(5L))
+ assert(master.sealRddChecksums(9) === 1) // partition 0 had no checksum
+ assert(master.getSealedChecksum(RDDBlockId(9, 1)) === Some(5L))
+ assert(master.getSealedChecksum(RDDBlockId(9, 0)).isEmpty)
+ }
+
+ test("sealed checksums are cleared when the RDD is removed") {
+ val store = makeBlockManager(20000, "exec1")
+ val blockId = RDDBlockId(10, 0)
+ master.updateBlockInfo(
+ store.blockManagerId, blockId, StorageLevel.DISK_ONLY, 0, 100, Some(7L))
+ master.sealRddChecksums(10)
+ assert(master.getSealedChecksum(blockId) === Some(7L))
+ master.removeRdd(10, blocking = true)
+ assert(master.getSealedChecksum(blockId).isEmpty)
+ }
+
+ test("local-checkpoint verification: agreeing disk and _SER replicas survive
the seal") {
+ val diskStore = makeBlockManager(20000, "exec1")
+ val serStore = makeBlockManager(20000, "exec2")
+ val blockId = RDDBlockId(11, 0)
+ val data = (1 to 32).toArray
+ // Same data on two executors at different serialized levels; both compute
a content checksum,
+ // which must agree across the disk and _SER store paths so the seal keeps
both replicas.
+ diskStore.getOrElseUpdateRDDBlock(
+ 1L, blockId, StorageLevel.DISK_ONLY, classTag[Int], () => data.iterator,
+ verifySealedChecksum = true)
+ serStore.getOrElseUpdateRDDBlock(
+ 2L, blockId, StorageLevel.MEMORY_ONLY_SER, classTag[Int], () =>
data.iterator,
+ verifySealedChecksum = true)
+ assert(master.getLocations(blockId).toSet ===
+ Set(diskStore.blockManagerId, serStore.blockManagerId))
+
+ assert(master.sealRddChecksums(11) === 0)
+ // A spurious format mismatch across paths would have evicted one replica.
+ assert(master.getLocations(blockId).toSet ===
+ Set(diskStore.blockManagerId, serStore.blockManagerId))
+ assert(master.getSealedChecksum(blockId).isDefined)
+ }
+
+ test("UpdateBlockInfo Externalizable round-trip preserves the checksum
field") {
+ def roundTrip(msg: UpdateBlockInfo): UpdateBlockInfo = {
+ val baos = new ByteArrayOutputStream()
+ val oos = new ObjectOutputStream(baos)
+ msg.writeExternal(oos)
+ oos.close()
+ val result = new UpdateBlockInfo()
+ result.readExternal(new ObjectInputStream(new
ByteArrayInputStream(baos.toByteArray)))
+ result
+ }
+ val bmId = BlockManagerId("exec", "host", 1234, None)
+ val withChecksum =
+ UpdateBlockInfo(bmId, RDDBlockId(1, 0), StorageLevel.DISK_ONLY, 0, 100,
Some(42L))
+ val withoutChecksum =
+ UpdateBlockInfo(bmId, RDDBlockId(1, 0), StorageLevel.DISK_ONLY, 0, 100,
None)
+ // The hand-rolled Externalizable must carry the checksum (the seal
depends on it reaching the
+ // master) and round-trip the other fields unchanged.
+ assert(roundTrip(withChecksum).checksum === Some(42L))
+ assert(roundTrip(withoutChecksum).checksum === None)
+ assert(roundTrip(withChecksum) === withChecksum)
+ }
+
+ test("divergent RDD block copies converge: seal evicts the minority and
reads skip it") {
+ val store1 = makeBlockManager(20000, "exec1")
+ val store2 = makeBlockManager(20000, "exec2")
+ val store3 = makeBlockManager(20000, "exec3")
+ val blockId = RDDBlockId(20, 0)
+ val dataA = Seq(1, 2, 3, 4)
+ val dataB = Seq(9, 9, 9, 9)
+ // Inject divergence: two executors materialize dataA, a third diverges
with dataB under the
+ // same block id (as a non-deterministic recompute by a speculative or
zombie attempt would).
+ store1.getOrElseUpdateRDDBlock(
+ 1L, blockId, StorageLevel.DISK_ONLY, classTag[Int],
+ () => dataA.iterator, verifySealedChecksum = true)
+ store3.getOrElseUpdateRDDBlock(
+ 3L, blockId, StorageLevel.DISK_ONLY, classTag[Int],
+ () => dataA.iterator, verifySealedChecksum = true)
+ store2.getOrElseUpdateRDDBlock(
+ 2L, blockId, StorageLevel.DISK_ONLY, classTag[Int],
+ () => dataB.iterator, verifySealedChecksum = true)
+
+ // The seal keeps the plurality (dataA, two copies) and evicts the
divergent dataB copy.
+ assert(master.sealRddChecksums(20) === 0)
+ assert(master.getLocations(blockId).toSet === Set(store1.blockManagerId,
store3.blockManagerId))
+ // The diverged executor no longer serves its stale local copy -- it was
evicted, or the
+ // read-side self-check skips it because its checksum != the sealed one
(so reads converge).
+ assert(store2.getLocalValues(blockId).isEmpty)
+ }
+
+ test("read-side self-check skips a divergent local copy of a sealed block
(no eviction)") {
+ // Isolates the read-side self-check from the seal's fire-and-forget
eviction: the block is the
+ // sole replica, so the seal keeps it (no RemoveBlock is sent), then we
make the local copy's
+ // checksum differ from the sealed value. getLocalValues must skip it
purely via the self-check.
+ val store = makeBlockManager(20000, "exec1")
+ val blockId = RDDBlockId(24, 0)
+ store.getOrElseUpdateRDDBlock(
+ 1L, blockId, StorageLevel.DISK_ONLY, classTag[Int], () => (1 to
16).iterator,
+ verifySealedChecksum = true)
+ assert(master.sealRddChecksums(24) === 0)
+ val sealedChecksum = master.getSealedChecksum(blockId)
+ assert(sealedChecksum.isDefined)
+ assert(store.getLocalValues(blockId).isDefined) // matches the seal, served
+
+ // Diverge the local copy from the sealed value (a stale replica the seal
is evicting), and
+ // clear the cached pull so the read re-consults the master.
+ val info = store.blockInfoManager.get(blockId).get
+ info.checksum = sealedChecksum.map(_ + 1)
+ info.sealedChecksum = None
+ // The self-check now skips the local copy; getLocalValues returns None
without any eviction.
+ assert(store.getLocalValues(blockId).isEmpty)
+ }
+
+ test("verifySealed tracks the seal request, not merely the presence of a
checksum") {
+ // A seal-path store (verifySealedChecksum = true) marks the block for the
read-side self-check.
+ val sealStore = makeBlockManager(20000, "exec1")
+ val sealBlock = RDDBlockId(21, 0)
+ sealStore.getOrElseUpdateRDDBlock(
+ 1L, sealBlock, StorageLevel.DISK_ONLY, classTag[Int], () => (1 to
16).iterator,
+ verifySealedChecksum = true)
+ val sealInfo = sealStore.blockInfoManager.get(sealBlock).get
+ assert(sealInfo.checksum.isDefined)
+ assert(sealInfo.verifySealedChecksum)
+
+ // A compute-only store (global flag on, no seal request) also gets a
checksum, but must NOT be
+ // marked for the self-check - having a checksum alone does not opt a
block in.
+ val checksumConf = new
SparkConf(false).set(STORAGE_RDD_BLOCK_CHECKSUM_ENABLED, true)
+ val computeStore = makeBlockManager(20000, "exec2", testConf =
Some(checksumConf))
+ val computeBlock = RDDBlockId(21, 1)
+ computeStore.getOrElseUpdateRDDBlock(
+ 2L, computeBlock, StorageLevel.DISK_ONLY, classTag[Int], () => (1 to
16).iterator)
+ val computeInfo = computeStore.blockInfoManager.get(computeBlock).get
+ assert(computeInfo.checksum.isDefined)
+ assert(!computeInfo.verifySealedChecksum)
+ }
+
+ test("global checksum flag records divergent replicas without evicting or
skipping them") {
+ // The compute-only mirror of the "divergent copies converge" seal test:
with the global flag
+ // on but no seal, divergent replicas are recorded but NOT converged -
nothing is evicted, no
+ // read is skipped, and no sealed checksum exists. Checksumming on its own
only observes.
+ val checksumConf = new
SparkConf(false).set(STORAGE_RDD_BLOCK_CHECKSUM_ENABLED, true)
+ val store1 = makeBlockManager(20000, "exec1", testConf =
Some(checksumConf))
+ val store2 = makeBlockManager(20000, "exec2", testConf =
Some(checksumConf))
+ val blockId = RDDBlockId(22, 0)
+ // No verifySealedChecksum flag: only the global switch drives the
checksum, so neither block is
+ // on the seal path.
+ store1.getOrElseUpdateRDDBlock(
+ 1L, blockId, StorageLevel.DISK_ONLY, classTag[Int], () => Seq(1, 2, 3,
4).iterator)
+ store2.getOrElseUpdateRDDBlock(
+ 2L, blockId, StorageLevel.DISK_ONLY, classTag[Int], () => Seq(9, 9, 9,
9).iterator)
+ // Both replicas stay in the directory (no seal => updateBlockInfo admits
both, divergent or
+ // not), and both are served locally - the read-side self-check does not
fire without a seal.
+ assert(master.getLocations(blockId).toSet === Set(store1.blockManagerId,
store2.blockManagerId))
+ assert(store1.getLocalValues(blockId).isDefined)
+ assert(store2.getLocalValues(blockId).isDefined)
+ assert(master.getSealedChecksum(blockId).isEmpty)
+
+ // The divergent checksums were recorded, so an explicit seal can still
converge them later
+ // (observe-then-enforce): the plurality is undefined at 1-vs-1, but
exactly one survives.
+ assert(master.sealRddChecksums(22) === 0)
+ assert(master.getLocations(blockId).size === 1)
+ assert(master.getSealedChecksum(blockId).isDefined)
+ }
+
+ test("BlockReplicationMetadata round-trips checksum and seal mark through
serialization") {
+ val ser = new JavaSerializer(conf).newInstance()
+ Seq(
+ BlockReplicationMetadata(StorageLevel.DISK_ONLY, classTag[Int]),
+ BlockReplicationMetadata(StorageLevel.DISK_ONLY, classTag[Int],
Some(42L), true)
+ ).foreach { meta =>
+ assert(ser.deserialize[BlockReplicationMetadata](ser.serialize(meta))
=== meta)
+ }
+ }
+
+ test("a seal-path RDD block replicates its checksum and mark, and the
replica is verifiable") {
+ val store1 = makeBlockManager(20000, "exec1")
+ val store2 = makeBlockManager(20000, "exec2")
+ val blockId = RDDBlockId(30, 0)
+ // Store a seal-path block with replication 2; the replica is uploaded to
a peer.
+ val level =
+ StorageLevel(useDisk = true, useMemory = false, deserialized = false,
replication = 2)
+ store1.getOrElseUpdateRDDBlock(
+ 1L, blockId, level, classTag[Int], () => Seq(1, 2, 3, 4).iterator,
+ verifySealedChecksum = true)
+ // Both the primary and the replica registered a checksum, and they agree
(same bytes).
+ assert(master.getLocations(blockId).size === 2)
+ assert(
+ master.sealRddChecksums(30) === 0, "both copies carried a checksum, so
none is unverified")
+ assert(master.getSealedChecksum(blockId).isDefined)
+ // Both locations survive the seal (they agreed), and the sealed block is
served from each.
+ assert(master.getLocations(blockId).size === 2)
+ assert(store1.getLocalValues(blockId).isDefined)
+ assert(store2.getLocalValues(blockId).isDefined)
+ }
test("SPARK-41497: mark rdd block as visible") {
val store = makeBlockManager(8000, "executor1")
@@ -2741,7 +3021,9 @@ class BlockManagerSuite extends SparkFunSuite with
Matchers with PrivateMethodTe
blockId: BlockId,
blockData: ManagedBuffer,
level: StorageLevel,
- classTag: ClassTag[_]): Future[Unit] = {
+ classTag: ClassTag[_],
+ checksum: Option[Long] = None,
+ verifySealedChecksum: Boolean = false): Future[Unit] = {
// scalastyle:off executioncontextglobal
import scala.concurrent.ExecutionContext.Implicits.global
// scalastyle:on executioncontextglobal
diff --git
a/core/src/test/scala/org/apache/spark/storage/FallbackStorageSuite.scala
b/core/src/test/scala/org/apache/spark/storage/FallbackStorageSuite.scala
index 6df8bc85b510..40fdd91207a1 100644
--- a/core/src/test/scala/org/apache/spark/storage/FallbackStorageSuite.scala
+++ b/core/src/test/scala/org/apache/spark/storage/FallbackStorageSuite.scala
@@ -243,7 +243,7 @@ class FallbackStorageSuite extends SparkFunSuite with
LocalSparkContext {
when(bm.master).thenReturn(bmm)
val blockTransferService = mock(classOf[BlockTransferService])
when(blockTransferService.uploadBlockSync(mc.any(), mc.any(), mc.any(),
mc.any(), mc.any(),
- mc.any(), mc.any())).thenThrow(new IOException)
+ mc.any(), mc.any(), mc.any(), mc.any())).thenThrow(new IOException)
when(bm.blockTransferService).thenReturn(blockTransferService)
when(bm.migratableResolver).thenReturn(resolver)
when(bm.getMigratableRDDBlocks()).thenReturn(Seq())
@@ -256,7 +256,8 @@ class FallbackStorageSuite extends SparkFunSuite with
LocalSparkContext {
eventually(timeout(10.second), interval(1.seconds)) {
// uploadBlockSync should not be used, verify that it is not called
verify(blockTransferService, never())
- .uploadBlockSync(mc.any(), mc.any(), mc.any(), mc.any(), mc.any(),
mc.any(), mc.any())
+ .uploadBlockSync(mc.any(), mc.any(), mc.any(), mc.any(), mc.any(),
mc.any(), mc.any(),
+ mc.any(), mc.any())
Seq("shuffle_1_1_0.index", "shuffle_1_1_0.data").foreach { filename =>
assert(fallbackStorage.exists(shuffleId = 1, filename))
diff --git
a/core/src/test/scala/org/apache/spark/storage/RddBlockChecksumBenchmark.scala
b/core/src/test/scala/org/apache/spark/storage/RddBlockChecksumBenchmark.scala
new file mode 100644
index 000000000000..2960c3d79a0c
--- /dev/null
+++
b/core/src/test/scala/org/apache/spark/storage/RddBlockChecksumBenchmark.scala
@@ -0,0 +1,180 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.storage
+
+import java.io.OutputStream
+
+import scala.reflect.ClassTag
+
+import org.apache.spark.SparkConf
+import org.apache.spark.benchmark.{Benchmark, BenchmarkBase}
+import org.apache.spark.internal.config
+import org.apache.spark.network.shuffle.checksum.ShuffleChecksumHelper
+import org.apache.spark.serializer.{KryoSerializer, SerializerManager}
+
+/**
+ * Benchmark for the store-time overhead of the local-checkpoint RDD-block
content checksum
+ * (`spark.checkpoint.local.verifyChecksum.enabled`). The feature folds a JDK
checksum
+ * over the serialized+compressed plaintext of each cache block as it is
written, by inserting a
+ * `MutableCheckedOutputStream` into the serialize sink chain
+ * (`ser.serializeStream(wrapForCompression(blockId, wrapForChecksum(checksum,
sink)))`, exactly as
+ * `BlockManager` composes it). This measures the cost of that inserted layer
on the block store
+ * path: for each data shape it times the exact serialize chain with and
without the checksum
+ * wrapper, so the delta is the per-block store-time tax the feature adds. A
final case times the
+ * raw checksum algorithms over a fixed buffer for reference.
+ *
+ * {{{
+ * To run this benchmark:
+ * 1. without sbt: bin/spark-submit --class <this class> <spark core test
jar>
+ * 2. build/sbt "core/Test/runMain <this class>"
+ * 3. generate result: SPARK_GENERATE_BENCHMARK_FILES=1 build/sbt
"core/Test/runMain <this class>"
+ * Results will be written to
"benchmarks/RddBlockChecksumBenchmark-results.txt".
+ * }}}
+ */
+object RddBlockChecksumBenchmark extends BenchmarkBase {
+
+ // Number of times each fixed-size block is serialized per iteration, so the
reported time is
+ // per-block-batch rather than dominated by per-call setup.
+ private val numBlocksPerIteration = 200
+
+ // Representative cache-block payloads. `longRows` are compact primitive
records; `byteRecords`
+ // are wider opaque byte payloads (the shape of a materialized DML source
row).
+ private def longRows(numRows: Int): Iterator[Long] = (0 until
numRows).iterator.map(_.toLong)
+ private def byteRecords(numRecords: Int, recordSize: Int):
Iterator[Array[Byte]] =
+ (0 until numRecords).iterator.map { i =>
+ val a = new Array[Byte](recordSize)
+ var j = 0
+ while (j < recordSize) {
+ a(j) = (i + j).toByte
+ j += 1
+ }
+ a
+ }
+
+ /**
+ * Serialize `values` for `blockId` through the exact store-time chain,
optionally inserting the
+ * checksum wrapper the feature adds. Mirrors
`SerializerManager.blockSerializationStream` and
+ * `BlockManager`'s `checksumOpt.map(wrapForChecksum(_,
sink)).getOrElse(sink)` composition
+ * (compression outside, checksum inside, sink at the bottom). `classTag` +
`autoPick = true`
+ * select the serializer exactly as the real store path does (a primitive /
byte-array block goes
+ * to Kryo); values are written as `Any` because the serializer instance,
not the static type,
+ * drives the work.
+ */
+ private def serializeBlock(
+ serManager: SerializerManager,
+ blockId: BlockId,
+ values: Iterator[Any],
+ withChecksum: Boolean,
+ algorithm: String,
+ classTag: ClassTag[_]): Unit = {
+ val sink: OutputStream = OutputStream.nullOutputStream()
+ val checksummed =
+ if (withChecksum) {
+
serManager.wrapForChecksum(ShuffleChecksumHelper.getChecksumByAlgorithm(algorithm),
sink)
+ } else {
+ sink
+ }
+ val compressed = serManager.wrapForCompression(blockId, checksummed)
+ val ser = serManager.getSerializer(classTag, autoPick = true).newInstance()
+ ser.serializeStream(compressed).writeAll(values).close()
+ }
+
+ private def newSerializerManager(compressRdds: Boolean): SerializerManager =
{
+ val conf = new SparkConf(false)
+ .set(config.RDD_COMPRESS, compressRdds)
+ // Kryo is the serializer DML source materialization uses in practice
(autoPick routes
+ // primitive / byte-array blocks to it), so match it here. Kryo's
FieldSerializer reflects into
+ // java.lang.invoke.SerializedLambda, which needs the `--add-opens
java.base/java.lang.invoke`
+ // JVM flag - supplied by the scala_binary target's jvm_flags (Spark's
COMMON_JDK17_FLAGS).
+ new SerializerManager(new KryoSerializer(conf), conf)
+ }
+
+ private def runStoreOverheadCase(
+ caseLabel: String,
+ compressRdds: Boolean,
+ makeValues: () => Iterator[Any],
+ classTag: ClassTag[_]): Unit = {
+ val serManager = newSerializerManager(compressRdds)
+ val blockId = RDDBlockId(0, 0)
+ val algorithm = "CRC32C"
+ val benchmark =
+ new Benchmark(caseLabel, numBlocksPerIteration.toLong, minNumIters = 5,
output = output)
+ benchmark.addCase("serialize only (feature off)") { _ =>
+ var i = 0
+ while (i < numBlocksPerIteration) {
+ serializeBlock(serManager, blockId, makeValues(), withChecksum =
false, algorithm, classTag)
+ i += 1
+ }
+ }
+ benchmark.addCase("serialize + CRC32C checksum (feature on)") { _ =>
+ var i = 0
+ while (i < numBlocksPerIteration) {
+ serializeBlock(serManager, blockId, makeValues(), withChecksum = true,
algorithm, classTag)
+ i += 1
+ }
+ }
+ // relativeTime shows the checksum case as a multiple of the
serialize-only baseline, i.e. the
+ // proportional store-time overhead the feature adds.
+ benchmark.run(relativeTime = true)
+ }
+
+ override def runBenchmarkSuite(mainArgs: Array[String]): Unit = {
+ runBenchmark("RDD block store-time checksum overhead") {
+ // 64k long rows (~512 KiB serialized) - a compact-record block.
+ runStoreOverheadCase(
+ "64k long rows, spark.rdd.compress=false",
+ compressRdds = false,
+ makeValues = () => longRows(64 * 1024),
+ classTag = implicitly[ClassTag[Long]])
+ runStoreOverheadCase(
+ "64k long rows, spark.rdd.compress=true",
+ compressRdds = true,
+ makeValues = () => longRows(64 * 1024),
+ classTag = implicitly[ClassTag[Long]])
+ // 8k x 128-byte records (~1 MiB serialized) - a wider-record block.
+ runStoreOverheadCase(
+ "8k x 128B records, spark.rdd.compress=false",
+ compressRdds = false,
+ makeValues = () => byteRecords(8 * 1024, 128),
+ classTag = implicitly[ClassTag[Array[Byte]]])
+ runStoreOverheadCase(
+ "8k x 128B records, spark.rdd.compress=true",
+ compressRdds = true,
+ makeValues = () => byteRecords(8 * 1024, 128),
+ classTag = implicitly[ClassTag[Array[Byte]]])
+ }
+
+ runBenchmark("Raw checksum algorithm over 4 MiB (reference)") {
+ val data: Array[Byte] = (0 until 4 * 1024 * 1024).map(_.toByte).toArray
+ val n = 256
+ val benchmark =
+ new Benchmark("Checksum algorithm", n.toLong, minNumIters = 5, output
= output)
+ Seq("ADLER32", "CRC32", "CRC32C").foreach { algorithm =>
+ benchmark.addCase(algorithm) { _ =>
+ var i = 0
+ while (i < n) {
+ val checksum =
ShuffleChecksumHelper.getChecksumByAlgorithm(algorithm)
+ checksum.update(data, 0, data.length)
+ i += 1
+ }
+ }
+ }
+ benchmark.run()
+ }
+ }
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
