Package com.linkedin.venice.spark.datawriter.task

Class SparkDataWriterTaskTracker

java.lang.Object

com.linkedin.venice.spark.datawriter.task.SparkDataWriterTaskTracker

All Implemented Interfaces:

DataWriterTaskTracker, TaskTracker

public class SparkDataWriterTaskTracker
extends Object
implements DataWriterTaskTracker

This class is used to track the metrics for the Spark Data Writer task.

Field Summary

Fields inherited from interface com.linkedin.venice.hadoop.task.TaskTracker

PROGRESS_COMPLETED, PROGRESS_NOT_SUPPORTED

Constructor Summary

Constructors

Constructor	Description
SparkDataWriterTaskTracker(DataWriterAccumulators accumulators)

Method Summary

Modifier and Type	Method	Description
long	getDuplicateKeyWithDistinctValueCount()
long	getExternalStorageWriteTimeMs()
Set<String>	getFailedExternalStorageRegions()	Returns the set of regions whose external writers exhausted retries and were reported by one or more data-writer tasks.
long	getIncrementalPushThrottledTimeMs()
int	getLargestUncompressedValueSize()	This accumulator performs a 'max' operation, which is not natively supported by either Spark or Hadoop.
long	getOutputRecordsCount()
long	getPartitionWriterCloseCount()
Map<Integer,Long>	getPerPartitionRecordCounts()	Returns per-partition record counts collected during the data writer job.
long	getRecordTooLargeFailureCount()
long	getRepushTtlFilterCount()
long	getSprayAllPartitionsCount()
long	getTotalGzipCompressedValueSize()
long	getTotalKeySize()
long	getTotalUncompressedValueSize()
long	getTotalValueSize()
long	getTotalZstdCompressedValueSize()
long	getUncompressedRecordTooLargeFailureCount()
long	getVeniceWriteTimeMs()
long	getWriteAclAuthorizationFailureCount()
void	setExternalStorageWriteTimeMs(long timeMs)	Sets the total external-storage write time summed on the driver from the successful task-output rows.
void	setFailedExternalStorageRegions(Set<String> failedRegions)	Sets the deduplicated failed external-storage regions collected from successful Spark task output.
void	setPerPartitionRecordCounts(Map<Integer,Long> counts)	Sets the per-partition record counts collected from the Spark DAG output via collect().
void	setVeniceWriteTimeMs(long timeMs)	Sets the total Venice write time summed on the driver from the successful task-output rows.
void	trackCompressedValueSize(int size)
void	trackDuplicateKeyWithDistinctValue(int count)
void	trackDuplicateKeyWithIdenticalValue(int count)
void	trackEmptyRecord()
void	trackExternalStorageWriteTime(long timeMs)	Report elapsed wall-clock time this task spent inside the external-storage write path.
void	trackFailedExternalStorageRegion(String regionName)	Report that the external writer for regionName exhausted its retry budget and was disabled for the remainder of the task.
void	trackGzipCompressedValueSize(int size)
void	trackIncrementalPushThrottledTime(long timeMs)
void	trackKeySize(int size)
void	trackLargestUncompressedValueSize(int size)	This accumulator performs a 'max' operation, which is not natively supported by either Spark or Hadoop.
void	trackPartitionWriterClose()
void	trackRecordSentToPubSub()
void	trackRecordTooLargeFailure()
void	trackRepushTtlFilteredRecord()
void	trackSprayAllPartitions()
void	trackUncompressedRecordTooLargeFailure()
void	trackUncompressedValueSize(int size)
void	trackVeniceWriteTime(long timeMs)	Report elapsed wall-clock time this task spent invoking the Venice/Kafka write path: the VeniceWriter.put invocations plus flushing and closing the Venice writer.
void	trackWriteAclAuthorizationFailure()
void	trackZstdCompressedValueSize(int size)

Methods inherited from class java.lang.Object

clone, equals, finalize, getClass, hashCode, notify, notifyAll, toString, wait, wait, wait

Methods inherited from interface com.linkedin.venice.hadoop.task.datawriter.DataWriterTaskTracker

getTotalPutOrDeleteRecordsCount, trackPutOrDeleteRecord

Methods inherited from interface com.linkedin.venice.hadoop.task.TaskTracker

getProgress, heartbeat

Constructor Details

SparkDataWriterTaskTracker

public SparkDataWriterTaskTracker(DataWriterAccumulators accumulators)

Method Details

trackSprayAllPartitions

public void trackSprayAllPartitions()

Specified by:

trackSprayAllPartitions in interface DataWriterTaskTracker

trackEmptyRecord

public void trackEmptyRecord()

Specified by:

trackEmptyRecord in interface DataWriterTaskTracker

trackKeySize

public void trackKeySize(int size)

Specified by:

trackKeySize in interface DataWriterTaskTracker

trackUncompressedValueSize

public void trackUncompressedValueSize(int size)

Specified by:

trackUncompressedValueSize in interface DataWriterTaskTracker

trackLargestUncompressedValueSize

public void trackLargestUncompressedValueSize(int size)

Description copied from interface: DataWriterTaskTracker

This accumulator performs a 'max' operation, which is not natively supported by either Spark or Hadoop. It is implemented using a custom accumulator in Spark.

Specified by:

trackLargestUncompressedValueSize in interface DataWriterTaskTracker

trackCompressedValueSize

public void trackCompressedValueSize(int size)

Specified by:

trackCompressedValueSize in interface DataWriterTaskTracker

trackGzipCompressedValueSize

public void trackGzipCompressedValueSize(int size)

Specified by:

trackGzipCompressedValueSize in interface DataWriterTaskTracker

trackZstdCompressedValueSize

public void trackZstdCompressedValueSize(int size)

Specified by:

trackZstdCompressedValueSize in interface DataWriterTaskTracker

trackWriteAclAuthorizationFailure

public void trackWriteAclAuthorizationFailure()

Specified by:

trackWriteAclAuthorizationFailure in interface DataWriterTaskTracker

trackRecordTooLargeFailure

public void trackRecordTooLargeFailure()

Specified by:

trackRecordTooLargeFailure in interface DataWriterTaskTracker

trackUncompressedRecordTooLargeFailure

public void trackUncompressedRecordTooLargeFailure()

Specified by:

trackUncompressedRecordTooLargeFailure in interface DataWriterTaskTracker

trackRecordSentToPubSub

public void trackRecordSentToPubSub()

Specified by:

trackRecordSentToPubSub in interface DataWriterTaskTracker

trackDuplicateKeyWithDistinctValue

public void trackDuplicateKeyWithDistinctValue(int count)

Specified by:

trackDuplicateKeyWithDistinctValue in interface DataWriterTaskTracker

trackDuplicateKeyWithIdenticalValue

public void trackDuplicateKeyWithIdenticalValue(int count)

Specified by:

trackDuplicateKeyWithIdenticalValue in interface DataWriterTaskTracker

trackRepushTtlFilteredRecord

public void trackRepushTtlFilteredRecord()

Specified by:

trackRepushTtlFilteredRecord in interface DataWriterTaskTracker

trackIncrementalPushThrottledTime

public void trackIncrementalPushThrottledTime(long timeMs)

Specified by:

trackIncrementalPushThrottledTime in interface DataWriterTaskTracker

trackFailedExternalStorageRegion

public void trackFailedExternalStorageRegion(String regionName)

Description copied from interface: DataWriterTaskTracker

Report that the external writer for regionName exhausted its retry budget and was disabled for the remainder of the task. Callers should report each region at most once per task.

Specified by:

trackFailedExternalStorageRegion in interface DataWriterTaskTracker

trackExternalStorageWriteTime

public void trackExternalStorageWriteTime(long timeMs)

Description copied from interface: DataWriterTaskTracker

Report elapsed wall-clock time this task spent inside the external-storage write path. This covers the complete external leg: the per-region write throttling wait, the ExternalStorageWriter.batchPut calls including their retries and retry backoff sleeps, the external flush, and the external close. It excludes anything spent producing to Venice/Kafka.

Callers report monotonic deltas as they accrue, so the aggregate is the sum of per-task durations, not the push's wall-clock duration: with N concurrent data-writer tasks the total can be up to N times the push's elapsed time.

Specified by:

trackExternalStorageWriteTime in interface DataWriterTaskTracker

trackVeniceWriteTime

public void trackVeniceWriteTime(long timeMs)

Description copied from interface: DataWriterTaskTracker

Report elapsed wall-clock time this task spent invoking the Venice/Kafka write path: the VeniceWriter.put invocations plus flushing and closing the Venice writer. It excludes anything spent in the external-storage path.

Same aggregation semantics as DataWriterTaskTracker.trackExternalStorageWriteTime(long): summed task durations, not push wall-clock time.

Specified by:

trackVeniceWriteTime in interface DataWriterTaskTracker

trackPartitionWriterClose

public void trackPartitionWriterClose()

Specified by:

trackPartitionWriterClose in interface DataWriterTaskTracker

getSprayAllPartitionsCount

public long getSprayAllPartitionsCount()

Specified by:

getSprayAllPartitionsCount in interface DataWriterTaskTracker

getTotalKeySize

public long getTotalKeySize()

Specified by:

getTotalKeySize in interface DataWriterTaskTracker

getTotalValueSize

public long getTotalValueSize()

Specified by:

getTotalValueSize in interface DataWriterTaskTracker

getTotalUncompressedValueSize

public long getTotalUncompressedValueSize()

Specified by:

getTotalUncompressedValueSize in interface DataWriterTaskTracker

getLargestUncompressedValueSize

public int getLargestUncompressedValueSize()

Description copied from interface: DataWriterTaskTracker

This accumulator performs a 'max' operation, which is not natively supported by either Spark or Hadoop. It is implemented using a custom accumulator in Spark.

Specified by:

getLargestUncompressedValueSize in interface DataWriterTaskTracker

getTotalGzipCompressedValueSize

public long getTotalGzipCompressedValueSize()

Specified by:

getTotalGzipCompressedValueSize in interface DataWriterTaskTracker

getTotalZstdCompressedValueSize

public long getTotalZstdCompressedValueSize()

Specified by:

getTotalZstdCompressedValueSize in interface DataWriterTaskTracker

getRecordTooLargeFailureCount

public long getRecordTooLargeFailureCount()

Specified by:

getRecordTooLargeFailureCount in interface DataWriterTaskTracker

getUncompressedRecordTooLargeFailureCount

public long getUncompressedRecordTooLargeFailureCount()

Specified by:

getUncompressedRecordTooLargeFailureCount in interface DataWriterTaskTracker

getWriteAclAuthorizationFailureCount

public long getWriteAclAuthorizationFailureCount()

Specified by:

getWriteAclAuthorizationFailureCount in interface DataWriterTaskTracker

getDuplicateKeyWithDistinctValueCount

public long getDuplicateKeyWithDistinctValueCount()

Specified by:

getDuplicateKeyWithDistinctValueCount in interface DataWriterTaskTracker

getOutputRecordsCount

public long getOutputRecordsCount()

Specified by:

getOutputRecordsCount in interface DataWriterTaskTracker

getPartitionWriterCloseCount

public long getPartitionWriterCloseCount()

Specified by:

getPartitionWriterCloseCount in interface DataWriterTaskTracker

getRepushTtlFilterCount

public long getRepushTtlFilterCount()

Specified by:

getRepushTtlFilterCount in interface DataWriterTaskTracker

getIncrementalPushThrottledTimeMs

public long getIncrementalPushThrottledTimeMs()

Specified by:

getIncrementalPushThrottledTimeMs in interface DataWriterTaskTracker

setPerPartitionRecordCounts

public void setPerPartitionRecordCounts(Map<Integer,Long> counts)

Sets the per-partition record counts collected from the Spark DAG output via collect().

setFailedExternalStorageRegions

public void setFailedExternalStorageRegions(Set<String> failedRegions)

Sets the deduplicated failed external-storage regions collected from successful Spark task output.

setExternalStorageWriteTimeMs

public void setExternalStorageWriteTimeMs(long timeMs)

Sets the total external-storage write time summed on the driver from the successful task-output rows.

setVeniceWriteTimeMs

public void setVeniceWriteTimeMs(long timeMs)

Sets the total Venice write time summed on the driver from the successful task-output rows.

getExternalStorageWriteTimeMs

public long getExternalStorageWriteTimeMs()

Specified by:

getExternalStorageWriteTimeMs in interface DataWriterTaskTracker

Returns:

the summed per-task duration (ms) spent in the external-storage write path across all successful task outputs of this push. See DataWriterTaskTracker.trackExternalStorageWriteTime(long) for exactly what is included; this is not the push's wall-clock time.

getVeniceWriteTimeMs

public long getVeniceWriteTimeMs()

Specified by:

getVeniceWriteTimeMs in interface DataWriterTaskTracker

Returns:

the summed per-task duration (ms) spent in the Venice/Kafka write path across all successful task outputs of this push. See DataWriterTaskTracker.trackVeniceWriteTime(long); this is not the push's wall-clock time.

getPerPartitionRecordCounts

public Map<Integer,Long> getPerPartitionRecordCounts()

Description copied from interface: DataWriterTaskTracker

Returns per-partition record counts collected during the data writer job. For the Spark path, these are collected via collect() on the DAG output

Specified by:

getPerPartitionRecordCounts in interface DataWriterTaskTracker

Returns:

Map of partition ID to record count, or empty map if not available.

getFailedExternalStorageRegions

public Set<String> getFailedExternalStorageRegions()

Description copied from interface: DataWriterTaskTracker

Returns the set of regions whose external writers exhausted retries and were reported by one or more data-writer tasks. Implementations should return an immutable or defensive-copy snapshot.

Specified by:

getFailedExternalStorageRegions in interface DataWriterTaskTracker