HoodieStreamer sample-writes record-size estimation fails with spark.rpc.message.maxSize for large records
- Dominant language
- Java
- Stars
- 6.2k
- Forks
- 2.5k
- Avg merge
- 2d 8h
- Merged PRs (30d)
- 111
Description
**Describe the problem you faced**
When `hoodie.streamer.sample.writes.enabled` is true (the default), HoodieStreamer estimates the average record size on the first commit by sampling up to `hoodie.streamer.sample.writes.size` records (default 25000) and writing them to a shadow table under `.hoodie/.aux/.sample_writes/`. The sampled records are collected to the driver and re-parallelized into a **single** Spark partition (`jsc.parallelize(samples, 1)`) for the sample bulk-insert.
For tables whose records have a large average size, the serialized single task exceeds `spark.rpc.message.maxSize` (default 128 MiB) and the sample write fails. The estimation is then skipped and the writer falls back to the default record-size estimate. Roughly, once `sample.writes.size × avg_serialized_record_bytes > spark.rpc.message.maxSize` (about > 5.4 KB per record at the defaults), the task can no longer be shipped to an executor.
**To Reproduce**
Steps to reproduce the behavior:
1. A new table (empty timeline) ingested via HoodieStreamer with `hoodie.streamer.sample.writes.enabled=true` (the default).
2. The first batch of records has a large average serialized size (> ~5.4 KB/record at default `sample.writes.size`).
3. The sample-writes record-size estimation fails; the sample-write task exceeds `spark.rpc.message.maxSize` and the estimation is skipped.
**Expected behavior**
Record-size estimation should not fail for large records. The sampler should stay within a size that can be shipped in a single task and still produce a usable average-record-size estimate.
**Environment Description**
* Hudi version : master (1.x)
* Spark version : 3.x
* Storage (HDFS/S3/GCS..) : any
* Running on Docker? (yes/no) : no
**Additional context**
The record-count cap (`hoodie.streamer.sample.writes.size`) does not bound the *serialized size* of the sample, which is what determines whether the single sample-write task fits under `spark.rpc.message.maxSize`.
Contributor guide
No contributing guide indexed for this repository
Assessment
This issue has not been assessed yet.