Repository navigation
feat(utilities): add HoodieSparkEventLogAnalyzer for Hudi-aware Spark stage analysis - #20196
Draft
nsivabalan wants to merge 2 commits into
Draft
nsivabalan wants to merge 2 commits into
nsivabalan wants to merge 2 commits into
Conversation
… stage analysis Summarises a Spark event log by Hudi operation and phase, so a stage report reads in Hudi's own vocabulary. A generic Spark profiler can say stage 14 was slow with 16000 tasks; it cannot say stage 14 was bloom-index probing inside an upsert, so its advice stays generic. This tool reads the job description Hudi publishes via HoodieEngineContext#setJobStatus and maps it onto a named operation and phase. The model is two levels, operation then phase, because a phase alone is ambiguous: a write inside an upsert is not the write inside a compaction, and forty repetitions of the rollback phases is one restore, not forty rollbacks. The operation is derived from the module half of the job description where it names one, otherwise from the phase sequence, and which signal was used is reported so a reader can judge confidence. Restore is detected by counting repeated rollbacks, because BaseRestoreActionExecutor sets no job status and simply calls the ordinary rollback path for each instant. Phases are deliberately coarse. The question worth answering is where the time went, at a granularity someone can act on: five buckets for a write operation, PLANNING and EXECUTION for every table service, and ARCHIVAL. The finer label is carried as a sub-phase so nothing is lost, and the verbatim job description is always reported as the escape hatch. "Building workload profile:" is the job under which deduplication, index tagging and profiling all run: Hudi sets the description once and the whole lazy chain executes beneath it. It therefore belongs to the tagging bucket rather than the write. The Spark stage name, which names the class that created the RDD, refines the sub-phase within that bucket so the composition stays visible -- a record index lookup and a bloom lookup are distinguishable there. The description always decides the phase; the stage name only adds detail, and is used to decide a phase only where there is no description at all. Parallel listings are attributed to the metadata table only when the listed paths are under it; a data-table listing is left unattributed rather than guessed at. Several bucket names state an attribution limit rather than hiding it. Hudi leaves the transformer, record creation and deduplication untagged, and Spark evaluates lazily, so their cost is forced into a neighbouring stage and cannot be separated from a log. SOURCE_READ_AND_TRANSFORM and DEDUP_AND_INDEX_TAGGING are named for everything they contain, and a caveats list names the buckets in each log that are prone to absorbing other work, so a downstream optimizer does not read a single share as ground truth. One limitation is called out rather than papered over: with hoodie.metadata.streaming.write.enabled, Hudi's default on Spark, the metadata write path runs with no job status at all and accounted for 61% of stage wall clock on a measured run. Those stages stay UNKNOWN, carry a sub-phase so a reader can see what the unattributed work is, and raise a caveat naming the config. The tool is accurate for non-streaming writes and 0.x-style DAGs and degrades on the 1.x streaming default. Both job-description encodings are supported, because logs of both are in circulation: the current "module:activity" in spark.job.description, and the pre-HUDI-8596 form where setJobGroup put the module in spark.jobGroup.id and the bare activity in the description. The tool is read-only and needs no SparkContext, so it runs as a plain java -cp entry point. Paths go through HoodieStorage, so s3/gcs/hdfs work. Parsing uses Jackson directly rather than Spark's JsonProtocol, which is private and version-coupled; unknown events are ignored, missing fields default, and a truncated final line is warned about rather than thrown on, since logs from killed applications routinely end mid-line. Output is TABLE for humans or JSON for machines. The JSON envelope is a stable contract carrying schemaVersion, application, operationSummary, phaseSummary, stages, caveats and warnings, with deterministic ordering; it is documented in HoodieSparkEventLogAnalyzer.md next to the tool, including the fact that Spark's output metrics are structurally zero for Hudi writes because Hudi writes Parquet through its own file handles. Exit codes are 0 on a successful parse and 2 for bad arguments or an unreadable log, including a bad flag.
Codecov Report❌ Patch coverage is Additional details and impacted files@@ Coverage Diff @@
## master #20196 +/- ##
============================================
+ Coverage 80.49% 80.79% +0.30%
- Complexity 34953 35610 +657
============================================
Files 2549 2570 +21
Lines 143060 145290 +2230
Branches 17392 18252 +860
============================================
+ Hits 115151 117383 +2232
+ Misses 20004 19852 -152
- Partials 7905 8055 +150
Flags with carried forward coverage won't be shown. Click here to find out more.
🚀 New features to boost your workflow:
|
3 tasks done
Collaborator
nsivabalan
marked this pull request as draft
October 8, 2026 03:09
… and add two attribution caveats - The analyzer opens no SparkContext but still needs Spark's jars on the classpath (Jackson among them); a bare `java -cp` on the utilities bundle fails with NoClassDefFoundError. The doc now shows spark-submit and says why. - New caveat under DEDUP_AND_INDEX_TAGGING: there is no simpleIndexLookup subPhase, because HoodieSimpleIndex builds its join through generic HoodieJavaRDD call sites; a stage whose planned task count equals the table's base-file count is the simple-index join regardless of its label. - New caveat under DATA_TABLE_WRITE: the stage that writes the data files is usually forced by the commit action and carries the commitStats subPhase; one task per written file group and most of the bucket's executor time identifies it as the write. Both observed on a replayed upsert against a 250-file table: three 250-task stages at 51% of stage wall clock labelled workloadProfile, and a 240-task, 70 s write stage labelled commitStats.
This branch has not been deployed
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
3 participants
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Describe the issue this Pull Request addresses
Part of #19262 (Hudi-aware agent skills and tools), under the Agentic Lakehouse umbrella #19256;
plan in #19262 (comment).
A user looking at a slow Hudi write sees Spark stages: "stage 14, 16000 tasks, 3 minutes". Spark
cannot tell them that stage 14 is bloom-index probing, that its parallelism comes from
hoodie.bloom.index.parallelism, or that the 10,000x task skew there means their key distributiondefeats range pruning. Generic Spark profilers have the stage metrics but no idea which part of Hudi
each stage belongs to, so their advice stays generic.
Hudi already labels its own Spark jobs —
HoodieSparkEngineContext.setJobStatussets a jobdescription on every significant step. This change reads those labels back out of a Spark event log
and turns a stage listing into a Hudi-phase breakdown.
Summary and Changelog
Adds
HoodieSparkEventLogAnalyzerplus a smalleventlogpackage. It parses a Spark event log(the file written when
spark.eventLog.enabled=true) and reports where a Hudi write spent its time,by Hudi phase and operation rather than by stage number.
No Spark session required — it reads a file, so it runs as a plain
java -cpprocess. Readsthrough
HoodieStorage, so local, S3 and GCS paths all work, as do rolling-log directories andgzip. The log is parsed with Jackson rather than Spark's
JsonProtocol, which is private andversion-coupled.
Phases. Nine buckets, deliberately coarse, because the useful answer is "60% of your time is
index tagging", not a per-stage taxonomy:
SOURCE_READ_AND_TRANSFORM,DEDUP_AND_INDEX_TAGGING,DATA_TABLE_WRITE,MARKER_RECONCILIATION,METADATA_TABLE_WRITEPLANNING,EXECUTIONARCHIVALandUNKNOWNTwo bucket names state an attribution limit rather than hiding it.
SOURCE_READ_AND_TRANSFORMisnamed for everything it contains because Hudi tags neither the transformer nor record creation, so
their cost is forced inside the source-read stage and cannot be separated.
DEDUP_AND_INDEX_TAGGINGlikewise covers one job: Hudi's
Building workload profile:is the description under which dedup,index tagging and profiling all execute.
A finer
subPhaseis kept alongside each stage —recordIndexLookup,bloomIndexLookup,smallFileProbe,commitStats,compactionPlanand others. That detail is what makes a coarsebucket actionable: a
DATA_TABLE_WRITErow at 57% says little, but the same row tagged[smallFileProbe]with a 10,000x skew points straight at a pathological partition.Operations. Thirteen values (
WRITE_OPERATION,COMPACTION,CLUSTERING,CLEANING,ROLLBACK,RESTORE,SAVEPOINT, …) resolved from the module half of the job description, fallingback to phase-sequence inference, then
UNKNOWN. The resolution source is reported so a reader canjudge it. This matters because
PLANNINGandEXECUTIONrecur across every table service — onlythe operation tells a compaction plan from a clustering plan. Repeated rollback groups are promoted
to
RESTORE, which is what a restore looks like from the outside.Two wire encodings. Before HUDI-8596,
setJobStatususedsetJobGroup, so the module landed inspark.jobGroup.idand the description held the bare activity. After it, both are combined into thedescription. Both are handled, so the tool works on pre-1.0 logs too.
Per stage: task count, duration percentiles, skew ratio, GC fraction, shuffle, spill, failure
reasons. Per application: critical path, peak concurrent executors, and the share of wall clock
per phase and per operation. TABLE output for humans,
--output JSONfor machines, with adocumented schema and deterministic ordering.
Caveats are part of the output, not a footnote. Hudi sets a job description before the Spark
action and Spark evaluates lazily, so a bucket names where work was forced, not where it logically
belongs. The JSON carries a
caveats[]array and the table output has a"How far to trust the attribution" section; entries fire only for the buckets actually present in
that log.
Known limitation, stated plainly in the docs. With
hoodie.metadata.streaming.write.enabled(the default on Spark), the streaming metadata-write path sets no job description at all, so a large
share of the DAG cannot be attributed. Measured on a local three-way comparison — same workload and
record-level index, toggling only that config — unattributed wall clock went from 3.9% to 60.5%. The
tool reports this honestly rather than guessing, and the breakdown is accurate for non-streaming
writes and for 0.x-style disjoint DAGs.
Also documented:
outputBytesandoutputRecordsare structurally zero for Hudi writes, becauseHudi writes Parquet through its own file handles and Spark never observes the bytes. Commit metadata
is the right source for written volume.
Tests —
TestHoodieSparkEventLogAnalyzer, 118 tests on synthetic event logs: both wireencodings, every phase, operation resolution by module and by sequence, rollback-to-restore
promotion, sub-phase refinement, percentile and skew arithmetic, zero-task stages, stage retries,
truncated and garbage logs, rolling directories, gzip, exit codes, and JSON determinism.
Validated end to end against real event logs from local Hudi writes across several metadata-table
configurations, and against a production Spark UI breakdown.
Impact
Purely additive. New class and new package in
hudi-utilities; no existing file is modified. Nopublic API change, no storage format change, no new
hoodie.*configs. Read-only: it parses a logfile and prints a report.
Risk Level
none
Documentation Update
HoodieSparkEventLogAnalyzer.mdcovers invocation, the phase and operation vocabularies, the JSONschema, and the caveats and limitations above.
Contributor's checklist