fix(streamer): bound sample-writes record-size sampling by serialized size - #19550
fix(streamer): bound sample-writes record-size sampling by serialized size#19550lokeshj1703 wants to merge 1 commit into
Conversation
hudi-agent
left a comment
hudi-agent
left a comment
There was a problem hiding this comment.
Thanks for working on this! This PR bounds the sample-writes record-size sampling by total serialized bytes (in addition to the existing record-count cap), moving the sampling onto the executor via mapPartitions on a single coalesced partition so the downstream single-task parallelize/bulkInsert stays under spark.rpc.message.maxSize. I traced the new takeBoundedSample logic (count vs. byte bounds, the always-retain-one guarantee, lazy early termination, and overflow), the resolveMaxSampleBytes derivation, and the downstream jsc.parallelize(samples, 1) write path, and everything holds up. No correctness issues found. A few style/readability suggestions in the inline comments. Please take a look, and this should be ready for a Hudi committer or PMC member to take it from here. One small naming inconsistency to address — MB vs MiB in a constant name — otherwise the code and tests are clean and well-documented.
cc @yihua
| /** Spark config bounding the maximum RPC frame size; a serialized task may not exceed it. */ | ||
| private static final String RPC_MESSAGE_MAX_SIZE_CONF = "spark.rpc.message.maxSize"; | ||
|
|
||
| /** Spark's default value for {@link #RPC_MESSAGE_MAX_SIZE_CONF}, in MiB (128 MiB). */ |
There was a problem hiding this comment.
🤖 nit: 1048576 bytes is one mebibyte (MiB), not one megabyte (MB = 1 000 000 bytes), and the Javadoc on the field directly above already says "in MiB". Could you rename this to BYTES_PER_MIB so the constant name stays consistent with the surrounding docs?
… size hoodie.streamer.sample.writes.enabled (default true) estimates average record size on the first commit by sampling up to hoodie.streamer.sample.writes.size records (default 25000) and shipping them to a single Spark task via jsc.parallelize(samples, 1). When the average record size is large, the serialized task exceeds spark.rpc.message.maxSize (default 128 MiB) and estimation fails, falling back to the default record-size estimate. Bound the sample by total serialized size in addition to record count. Sampling now runs on the executor via mapPartitions on the single coalesced partition and stops once the accumulated serialized size would exceed 0.5 of spark.rpc.message.maxSize, so neither the driver collect nor the downstream single-partition write can exceed the RPC frame limit. At least one record is always retained. Behavior is unchanged where the sample already fits under the budget.
42034af to
f470f00
Compare
Codecov Report✅ All modified and coverable lines are covered by tests. Additional details and impacted files@@ Coverage Diff @@
## master #19550 +/- ##
============================================
- Coverage 76.84% 76.83% -0.01%
+ Complexity 32379 32377 -2
============================================
Files 2522 2522
Lines 139106 139122 +16
Branches 16713 16715 +2
============================================
- Hits 106892 106891 -1
- Misses 24621 24636 +15
- Partials 7593 7595 +2
Flags with carried forward coverage won't be shown. Click here to find out more.
🚀 New features to boost your workflow:
|
hudi-agent
left a comment
hudi-agent
left a comment
There was a problem hiding this comment.
Thanks for working on this! This PR bounds the sample-writes record-size sampling by total serialized size (0.5 × spark.rpc.message.maxSize) in addition to record count, moving the sampling onto the executor via mapPartitions so the single-task sample write stays under the RPC frame limit. I traced the count/byte bounds, the always-retain-one guarantee, the RPC-config resolution, and the executor-to-driver shipping path, and didn't find correctness issues in this round — notably, if the byte budget's headroom ever proved insufficient the change degrades to the pre-PR fallback rather than corrupting anything, so there's no regression risk. No new issues flagged from this automated pass — a Hudi committer or PMC member can take it from here for a final review. A single minor naming suggestion below; the fix is otherwise clean and well-documented.
cc @yihua
| break; | ||
| } | ||
| samples.add(sample); | ||
| accumulatedBytes += recordBytes; |
There was a problem hiding this comment.
🤖 nit: the "" literal carries the semantics "write to a single non-partitioned file", which the Javadoc explains but the call-site doesn't. Could you pull it out as a class-level constant (e.g. SAMPLE_WRITES_PARTITION_PATH = "") so both this line and the test's serializedSizeOfFirst helper point to the same named thing rather than two independent empty strings?
4 participants
Describe the issue this Pull Request addresses
Closes #19549.
hoodie.streamer.sample.writes.enabled(default true) estimates average record size on the first commit by sampling up tohoodie.streamer.sample.writes.sizerecords (default 25000) and shipping them to a single Spark task viajsc.parallelize(samples, 1). When the average record size is large, the serialized task exceedsspark.rpc.message.maxSize(default 128 MiB) and estimation fails, falling back to the default record-size estimate.Summary and Changelog
Bound the sample by total serialized size in addition to record count:
mapPartitionson the single coalesced partition and stops once the accumulated serialized size would exceed0.5 × spark.rpc.message.maxSize. This keeps both the driver collect and the downstream single-partition write under the RPC frame limit.SerializationUtils.serialize(...).length, which reflects the bytes actually shipped in the task (for the Avro record type the payload holdsrecordBytes, so the schema is not counted per record).hoodie.streamer.sample.writes.size) still applies; the byte cap only engages where the previous code would have failed, so behavior is unchanged for tables whose sample already fits under the budget.Added
TestSparkSampleWritesBoundingcovering the count bound, the byte-budget boundary, monotonicity of the sample size with the budget, and the always-retain-one guarantee.Impact
No config or public API change. For large-record tables the first-commit record-size estimate now succeeds instead of silently falling back to the default estimate; subsequent commits already derive record size from real commit metadata.
Risk Level
low
Sampling-only change. Existing sample-writes tests are unchanged and new unit tests were added for the bounding logic.
Documentation Update
none
Contributor's checklist
🤖 Generated with Claude Code