Skip to content

Add named accumulators to track PerfIO S3 backend usage per executor #14580

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
zpuller merged 4 commits into from
Apr 17, 2026
Merged

Add named accumulators to track PerfIO S3 backend usage per executor #14580

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
654b419
perfio accumulator
zpuller Apr 9, 2026
1563853
fix
zpuller Apr 16, 2026
aa6ab1d
add recordPerfioS3BackendOnce to copyRemoteBlocksData
zpuller Apr 16, 2026
c550e5e
catch accumulator not yet registered for task
zpuller Apr 16, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
12 changes: 9 additions & 3 deletions sql-plugin/src/main/scala/com/nvidia/spark/rapids/parquet/GpuParquetScan.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -78,8 +78,8 @@ import org.apache.spark.sql.execution.datasources.{DataSourceUtils, PartitionedF
import org.apache.spark.sql.execution.datasources.v2.FileScan
import org.apache.spark.sql.execution.datasources.v2.parquet.ParquetScan
import org.apache.spark.sql.internal.SQLConf
import org.apache.spark.sql.rapids.{isTimestampNTZ, GpuTaskMetrics}
import org.apache.spark.sql.rapids.execution.TrampolineUtil
import org.apache.spark.sql.rapids.isTimestampNTZ
import org.apache.spark.sql.sources.Filter
import org.apache.spark.sql.types._
import org.apache.spark.sql.util.CaseInsensitiveStringMap
Expand Down Expand Up @@ -565,8 +565,11 @@ protected case class GpuParquetFileFilterHandler(
if (fileIO.isInstanceOf[HadoopFileIO]) {
// We should remove this after https://github.com/NVIDIA/spark-rapids/issues/13306 is
// implemented.
PerfIO.readParquetFooterBuffer(filePath, conf, verifyParquetMagic)
.getOrElse(readFooterBufUsingHadoop(fileIO, filePath))
val result = PerfIO.readParquetFooterBuffer(filePath, conf, verifyParquetMagic)
if (filePath.toUri.getScheme.startsWith("s3")) {
GpuTaskMetrics.get.recordPerfioS3BackendOnce()
}
Comment thread
zpuller marked this conversation as resolved.
result.getOrElse(readFooterBufUsingHadoop(fileIO, filePath))
} else {
readFooterBufUsingHadoop(fileIO, filePath)
}
Expand Down Expand Up @@ -2036,6 +2039,9 @@ trait ParquetPartitionReaderBase extends Logging with ScanWithMetrics

val totalBytesCopied = if (fileIO.isInstanceOf[HadoopFileIO]) {
// Fix this after https://github.com/NVIDIA/spark-rapids/issues/13306 is resolved
if (filePath.toUri.getScheme.startsWith("s3")) {
GpuTaskMetrics.get.recordPerfioS3BackendOnce()
}
PerfIO.readToHostMemory(
conf, out.buffer, filePath.toUri,
coalescedRanges.map(r => IntRangeWithOffset(r.offset, r.length, r.outputOffset))
Expand Down
33 changes: 31 additions & 2 deletions sql-plugin/src/main/scala/org/apache/spark/sql/rapids/GpuTaskMetrics.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -24,7 +24,7 @@ import java.util.concurrent.TimeUnit
import java.util.concurrent.atomic.AtomicLong

import ai.rapids.cudf.{NvtxColor, NvtxRange}
import com.nvidia.spark.rapids.{NvtxId, NvtxRegistry}
import com.nvidia.spark.rapids.{NvtxId, NvtxRegistry, PerfIO}
import com.nvidia.spark.rapids.Arm.withResource
import com.nvidia.spark.rapids.ScalableTaskCompletion.onTaskCompletion
import com.nvidia.spark.rapids.jni.RmmSpark
Expand Down Expand Up @@ -281,6 +281,12 @@ class GpuTaskMetrics extends Serializable with Logging {
// Disk write savings from SpillablePartialFileHandle
private val diskWriteSavedBytes = new LongAccumulator

// PerfIO S3 backend executor counts — each executor contributes 1 to its active backend
// per stage, making it easy to see from the event log whether PerfIO is enabled on all nodes.
private val perfioS3NettyExecutors = new LongAccumulator
private val perfioS3CrtExecutors = new LongAccumulator
private val perfioS3S3aExecutors = new LongAccumulator

private var maxHostBytesAllocated: Long = 0
private var maxPageableBytesAllocated: Long = 0
private var maxPinnedBytesAllocated: Long = 0
Expand Down Expand Up @@ -343,7 +349,10 @@ class GpuTaskMetrics extends Serializable with Logging {
"gpuMaxTaskFootprint" -> maxGpuFootprint,
"multithreadReaderMaxParallelism" -> multithreadReaderMaxParallelism,
"gpuMaxConcurrentGpuTasks" -> maxConcurrentGpuTasks,
"gpuDiskWriteSavedBytes" -> diskWriteSavedBytes
"gpuDiskWriteSavedBytes" -> diskWriteSavedBytes,
"perfio.s3.netty.executors" -> perfioS3NettyExecutors,
"perfio.s3.crt.executors" -> perfioS3CrtExecutors,
"perfio.s3.s3a.executors" -> perfioS3S3aExecutors
)

def register(sc: SparkContext): Unit = {
Expand Down Expand Up @@ -497,6 +506,26 @@ class GpuTaskMetrics extends Serializable with Logging {
def addDiskWriteSaved(bytes: Long): Unit = {
diskWriteSavedBytes.add(bytes)
}

/**
* Records this executor's PerfIO S3 backend exactly once per stage (per GpuTaskMetrics
* instance). Call from task code on any S3 read path. Uses the accumulator ID as a key
* to prevent double-counting — each new stage creates fresh accumulators with new IDs.
*/
def recordPerfioS3BackendOnce(): Unit = {
val acc = PerfIO.s3BackendName match {
case "netty" => perfioS3NettyExecutors
case "crt" => perfioS3CrtExecutors
case _ => perfioS3S3aExecutors
}
try {
if (PerfIO.reportedBackendAccIds.add(acc.id)) {
acc.add(1L)
}
} catch {
case _: IllegalArgumentException => // accumulator not yet registered; no-op
}
}
}

/**
Expand Down