ShuffleReadMetricsReporter

Author: jaceklaskowski

docs/configuration-properties.md

@@ -629,7 +629,7 @@ Must be a positive integer (above `0`)
 
 **spark.shuffle.push.enabled**
 
-Enables push-based shuffle on the client side
+Enables [Push-Based Shuffle](./push-based-shuffle.md) on the client side
 
 Default: `false`
 

docs/shuffle/BlockStoreShuffleReader.md

@@ -1,6 +1,6 @@
 # BlockStoreShuffleReader
 
-`BlockStoreShuffleReader` is a [ShuffleReader](ShuffleReader.md).
+`BlockStoreShuffleReader[K, C]` is a [ShuffleReader](ShuffleReader.md) of `K` keys and `C` values.
 
 ## Creating Instance
 
@@ -9,7 +9,7 @@
 * <span id="handle"> [BaseShuffleHandle](BaseShuffleHandle.md)
 * <span id="blocksByAddress"> [Block](../storage/BlockId.md)s by [Address](../storage/BlockManagerId.md) (`Iterator[(BlockManagerId, Seq[(BlockId, Long, Int)])]`)
 * <span id="context"> [TaskContext](../scheduler/TaskContext.md)
-* <span id="readMetrics"> `ShuffleReadMetricsReporter`
+* <span id="readMetrics"> [ShuffleReadMetricsReporter](ShuffleReadMetricsReporter.md)
 * <span id="serializerManager"> [SerializerManager](../serializer/SerializerManager.md)
 * <span id="blockManager"> [BlockManager](../storage/BlockManager.md)
 * <span id="mapOutputTracker"> [MapOutputTracker](../scheduler/MapOutputTracker.md)

docs/shuffle/ShuffleReadMetricsReporter.md

@@ -0,0 +1,33 @@
+# ShuffleReadMetricsReporter
+
+`ShuffleReadMetricsReporter` is an abstraction of [reporters](#implementations) that allow tracking the following **Shuffle Read Metrics** (for each shuffle):
+
+Shuffle Read Metric | When used
+-|-
+ Corrupt Merged Block Chunks | FIXME
+ Fetch Wait Time | FIXME
+ Local Blocks Fetched | FIXME
+ Local Bytes Read | FIXME
+ Merged Local Blocks Fetched | FIXME
+ Merged Local Bytes Read | FIXME
+ Merged Local Chunks Fetched | FIXME
+ Merged Fetch Fallback Count | FIXME
+ Merged Remote Blocks Fetched | FIXME
+ Merged Remote Chunks Fetched | FIXME
+ Merged Remote Bytes Read | FIXME
+ Merged Remote Requests Duration | FIXME
+ Remote Blocks Fetched | [ShuffleBlockFetcherIterator](../storage/ShuffleBlockFetcherIterator.md#shuffleRemoteMetricsUpdate) for [ShuffleBlockChunkId](../storage/BlockId.md#ShuffleBlockChunkId)s
+ Remote Bytes Read | FIXME
+ Remote Bytes Read To Disk | FIXME
+ Remote Requests Duration | FIXME
+ Total Records Read | FIXME
+
+`ShuffleReadMetricsReporter` is used to create the following:
+
+* [BlockStoreShuffleReader](BlockStoreShuffleReader.md#readMetrics)
+* `PushBasedFetchHelper`
+* [ShuffleBlockFetcherIterator](../storage/ShuffleBlockFetcherIterator.md#shuffleMetrics)
+
+## Implementations
+
+* `TempShuffleReadMetrics`

docs/storage/BlockId.md

@@ -4,7 +4,7 @@
 
 ## Contract
 
-### <span id="name"><span id="toString"> Name
+### <span id="toString"> Name { #name }
 
 ```scala
 name: String
@@ -24,7 +24,9 @@ Used when:
 ??? note "Sealed Abstract Class"
     `BlockId` is a Scala **sealed abstract class** which means that all of the implementations are in the same compilation unit (a single file).
 
-### <span id="BroadcastBlockId"> BroadcastBlockId
+    Learn more in the [Scala Language Specification]({{ scala.spec }}/05-classes-and-objects.html#sealed).
+
+### BroadcastBlockId { #BroadcastBlockId }
 
 `BlockId` for [broadcast variable](../broadcast-variables/index.md) blocks:
 
@@ -40,7 +42,15 @@ Used when:
 * `SerializerManager` is requested to [shouldCompress](../serializer/SerializerManager.md#shouldCompress)
 * `AppStatusListener` is requested to [onBlockUpdated](../status/AppStatusListener.md#onBlockUpdated)
 
-### <span id="RDDBlockId"> RDDBlockId
+### CacheId { #CacheId }
+
+`BlockId` for...FIXME
+
+### PythonStreamBlockId { #PythonStreamBlockId }
+
+`BlockId` for...FIXME
+
+### RDDBlockId { #RDDBlockId }
 
 `BlockId` for [RDD](../rdd/RDD.md) partitions:
 
@@ -60,9 +70,22 @@ Used when:
 
 [Compressed](../serializer/SerializerManager.md#shouldCompress) when [spark.rdd.compress](../configuration-properties.md#spark.rdd.compress) configuration property is enabled
 
-### <span id="ShuffleBlockBatchId"> ShuffleBlockBatchId
+### ShuffleBlockBatchId { #ShuffleBlockBatchId }
+
+`BlockId` for...FIXME
 
-### <span id="ShuffleBlockId"> ShuffleBlockId
+### ShuffleBlockChunkId { #ShuffleBlockChunkId }
+
+`BlockId` for shuffle block chunks in [Push-Based Shuffle](../push-based-shuffle.md):
+
+* `shuffleId` identifier
+* `shuffleMergeId` identifier
+* `reduceId` identifier
+* `chunkId` identifier
+
+Uses `shuffleChunk_[shuffleId]_[shuffleMergeId]_[reduceId]_[chunkId]` pattern for the [name](#name)
+
+### ShuffleBlockId { #ShuffleBlockId }
 
 `BlockId` for shuffle blocks:
 
@@ -83,13 +106,41 @@ Used when:
 
 [Compressed](../serializer/SerializerManager.md#shouldCompress) when [spark.shuffle.compress](../configuration-properties.md#spark.shuffle.compress) configuration property is enabled
 
-### <span id="ShuffleDataBlockId"> ShuffleDataBlockId
+### ShuffleChecksumBlockId { #ShuffleChecksumBlockId }
+
+`BlockId` for...FIXME
+
+### ShuffleDataBlockId { #ShuffleDataBlockId }
+
+`BlockId` for...FIXME
 
-### <span id="ShuffleIndexBlockId"> ShuffleIndexBlockId
+### ShuffleIndexBlockId { #ShuffleIndexBlockId }
 
-### <span id="StreamBlockId"> StreamBlockId
+`BlockId` for...FIXME
 
-`BlockId` for ...FIXME:
+### ShuffleMergedBlockId { #ShuffleMergedBlockId }
+
+`BlockId` for...FIXME
+
+### ShuffleMergedDataBlockId { #ShuffleMergedDataBlockId }
+
+`BlockId` for...FIXME
+
+### ShuffleMergedIndexBlockId { #ShuffleMergedIndexBlockId }
+
+`BlockId` for...FIXME
+
+### ShuffleMergedMetaBlockId { #ShuffleMergedMetaBlockId }
+
+`BlockId` for...FIXME
+
+### ShufflePushBlockId { #ShufflePushBlockId }
+
+`BlockId` for...FIXME
+
+### StreamBlockId { #StreamBlockId }
+
+`BlockId` for...FIXME:
 
 * `streamId`
 * `uniqueId`
@@ -102,15 +153,23 @@ input-[streamId]-[uniqueId]
 
 Used in Spark Streaming
 
-### <span id="TaskResultBlockId"> TaskResultBlockId
+### TaskResultBlockId { #TaskResultBlockId }
+
+`BlockId` for...FIXME
+
+### TempLocalBlockId { #TempLocalBlockId }
+
+`BlockId` for...FIXME
+
+### TempShuffleBlockId { #TempShuffleBlockId }
 
-### <span id="TempLocalBlockId"> TempLocalBlockId
+`BlockId` for...FIXME
 
-### <span id="TempShuffleBlockId"> TempShuffleBlockId
+### TestBlockId { #TestBlockId }
 
-### <span id="TestBlockId"> TestBlockId
+`BlockId` for...FIXME
 
-## <span id="apply"> Creating BlockId by Name
+## Creating BlockId by Name { #apply }
 
 ```scala
 apply(

docs/storage/ShuffleBlockFetcherIterator.md

@@ -23,7 +23,7 @@
 * <span id="maxReqSizeShuffleToMem"> [spark.network.maxRemoteBlockSizeFetchToMem](../configuration-properties.md#spark.network.maxRemoteBlockSizeFetchToMem)
 * <span id="detectCorrupt"> [spark.shuffle.detectCorrupt](../configuration-properties.md#spark.shuffle.detectCorrupt)
 * <span id="detectCorruptUseExtraMemory"> [spark.shuffle.detectCorrupt.useExtraMemory](../configuration-properties.md#spark.shuffle.detectCorrupt.useExtraMemory)
-* <span id="shuffleMetrics"> `ShuffleReadMetricsReporter`
+* <span id="shuffleMetrics"> [ShuffleReadMetricsReporter](../shuffle/ShuffleReadMetricsReporter.md)
 * <span id="doBatchFetch"> `doBatchFetch` flag
 
 While being created, `ShuffleBlockFetcherIterator` [initializes itself](#initialize).
@@ -185,7 +185,7 @@ next(): (BlockId, InputStream)
 
 `next` takes (and removes) the head of the [results](#results) queue.
 
-`next` requests the [ShuffleReadMetricsReporter](#shuffleMetrics) to `incFetchWaitTime`.
+`next` requests the [ShuffleReadMetricsReporter](#shuffleMetrics) to [incFetchWaitTime](../shuffle/ShuffleReadMetricsReporter.md#incFetchWaitTime).
 
 `next`...FIXME
 

mkdocs.yml

@@ -184,6 +184,7 @@ nav:
       - ShuffleExternalSorter: shuffle/ShuffleExternalSorter.md
       - ShuffleInMemorySorter: shuffle/ShuffleInMemorySorter.md
       - ShuffleManager: shuffle/ShuffleManager.md
+      - shuffle/ShuffleReadMetricsReporter.md
       - ShuffleWriteMetricsReporter: shuffle/ShuffleWriteMetricsReporter.md
       - ShuffleWriteProcessor: shuffle/ShuffleWriteProcessor.md
       - ShuffleWriter: shuffle/ShuffleWriter.md