ntopousis opened a new issue, #39370:
URL: https://github.com/apache/beam/issues/39370
🤖 This report and proposed patch were prepared with an AI coding agent from
a production investigation. Identifiers and application-specific details are
intentionally omitted; the Dataflow job and support-case identifiers can be
shared privately with the Google Dataflow team.
### What happened?
A bounded Java pipeline using Apache Beam 2.64.0 on Google Cloud Dataflow
took multiple hours to finalize several otherwise-completed unwindowed
`TextIO.write()` transforms using runner-determined sharding.
The distinguishing factor was not side-input byte size. It was the number of
temporary-file metadata records entering `WriteFiles/GatherTempFileResults`.
| Output | `FileResult` count | Observed status |
|---|---:|---|
| A | 5,119 | completed |
| B | 23,633 | completed |
| C | 47,285 | completed |
| D | 89,677 | still finalizing |
| E | 97,106 | still finalizing |
One corresponding internal side-input materialization was approximately 18
MiB. The two largest finalizers were not deadlocked: they advanced
near-linearly at roughly 200–225 `FileResult` records per minute, making
finalization take many hours.
Google Cloud support's backend analysis reported that the job was spending
substantial time reading and retrying GCS files. Repeated worker stacks during
this phase included:
```text
GoogleCloudStorageReadChannel
-> IsmReaderImpl
-> IsmSideInputReader
-> IterableLikeCoder.registerByteSizeObserver
-> OutputObjectAndByteCounter
```
This was not an application-created `PCollectionView`. It was the internal
side input introduced by bounded, unwindowed `WriteFiles`.
We have not proved whether the dominant defect is the SDK's use of a global
side input, Dataflow's access pattern for a high-cardinality ISM iterable, or
their interaction. The proposed SDK patch below removes that interaction while
deliberately leaving finalization semantics unchanged.
### Relevant SDK path
Both fixed sharding and bounded runner-determined sharding feed their
`FileResult`s into `GatherTempFileResults`:
https://github.com/apache/beam/blob/v2.64.0/sdks/java/core/src/main/java/org/apache/beam/sdk/io/WriteFiles.java#L469-L490
For unwindowed writes, `GatherResults` reshuffles the records, creates
`View.asIterable()`, reifies the view in the global window, and copies the
iterable into a `List`:
https://github.com/apache/beam/blob/v2.64.0/sdks/java/core/src/main/java/org/apache/beam/sdk/io/WriteFiles.java#L527-L570
`FinalizeFn` then copies that list again before finalizing all destinations
and moving the files:
https://github.com/apache/beam/blob/v2.64.0/sdks/java/core/src/main/java/org/apache/beam/sdk/io/WriteFiles.java#L1335-L1384
The same global side-input path remains on master as of
`7eef1304b19c5a47c12354476c4319df5183908d`:
https://github.com/apache/beam/blob/7eef1304b19c5a47c12354476c4319df5183908d/sdks/java/core/src/main/java/org/apache/beam/sdk/io/WriteFiles.java#L534-L577
PR #26289 introduced `View.asIterable()` in place of `View.asList()` to
reduce the cost of small writes, while preserving a downstream `ListCoder` for
update compatibility:
https://github.com/apache/beam/pull/26289
This report concerns the high-cardinality failure mode that remains in that
path.
### Reproduction shape
The production trigger can be approximated with a bounded pipeline that
induces tens of thousands of runner bundles and then performs an unwindowed
write without explicit sharding:
```java
PCollection<String> records =
pipeline
.apply(GenerateSequence.from(0).to(NUM_RECORDS))
.apply(/* produce fixed-size records */)
.apply(Reshuffle.viaRandomKey());
records.apply(TextIO.write().to("gs://BUCKET/reproduction/output"));
// Intentionally no withNumShards() and no withWindowedWrites().
```
The exact `FileResult` count is runner-dependent, so the meaningful
reproduction condition is the cardinality entering `GatherTempFileResults`, not
only the number of input records.
### Expected behavior
Finalization necessarily performs work proportional to the number of
temporary files, but an approximately 18 MiB metadata collection should not
take hours to consume. The gather should use a normal shuffle or another bulk
main-input path rather than a remote iterable side-input access pattern with
very high per-element overhead.
### Proposed narrow SDK patch
The accompanying draft patch changes only the bounded, unwindowed gather
transport:
1. Gather `FileResult`s into bundle-sized lists.
2. Add one explicitly coded empty-list marker so empty writes still reach
finalization.
3. Flatten and key those lists by `Void`.
4. Use `GroupByKey` as a main-input shuffle/checkpoint.
5. Flatten the grouped bundle lists back into the existing
`PCollection<List<FileResult<...>>>` contract.
6. Avoid the redundant copy in `FinalizeFn`.
The patch retains `ListCoder` and leaves the existing finalizer and
`FileBasedSink.WriteOperation` behavior intact. It therefore preserves:
- empty unwindowed writes producing at least one shard unless `skipIfEmpty`
is enabled;
- missing empty-shard creation for fixed sharding;
- runner-determined shard numbering and final filename semantics;
- dynamic destinations and filename-policy side inputs;
- retry-safe rename options;
- output-filename visibility after finalization;
- whole-temporary-directory cleanup after all renames.
A structural regression test asserts that the unwindowed
`GatherTempFileResults` subtree no longer contains
`View.CreatePCollectionView`. Existing `WriteFilesTest` coverage exercises
empty writes, fixed sharding, missing shards, dynamic destinations, filenames,
and cleanup.
### Deliberate limitation
This is a narrow mitigation, not a complete finalization redesign. It still
produces one O(N) list and uses one logical finalizer. Truly memory-bounded,
parallel finalization would require a two-phase design because shard counts and
final names must be established before parallel renames, missing empty shards
must be generated exactly once, and the shared unwindowed temporary directory
cannot be deleted until every rename chunk completes.
A Dataflow-side improvement to sequential ISM reads, range prefetch, or
coder byte-size observation may also be appropriate to protect pipelines on
released SDK versions.
### Questions for maintainers
1. Is the global side-input gather intentional for any reason beyond
ensuring an empty write still invokes finalization?
2. Is the empty-marker plus main-input `GroupByKey` approach acceptable as a
narrow first fix?
3. Does the Dataflow team recognize this `IsmReaderImpl` /
`IterableLikeCoder.registerByteSizeObserver` stack as a runner-side
read-amplification pattern?
4. Should chunked/parallel finalization be tracked as a separate follow-up
after removing the side-input bottleneck?
### Issue Priority
Priority: 2 (default / most bugs should be filed as P2)
### Issue Components
- [x] Component: Java SDK
- [x] Component: IO connector
- [x] Component: Google Cloud Dataflow Runner
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
