jonathansudhakar1 opened a new pull request, #41318:
URL: https://github.com/apache/beam/pull/41318
Fixes #25769
Fixes #41245
**What**
1. **Docs (#25769).** The class docstrings of `WriteToText`, `WriteToAvro`,
`WriteToParquet`, `WriteToParquetBatched` and `WriteToTFRecord` now say that
the output is a `PCollection` of `str` holding the final path of each file
written, one per shard, emitted once that file is in place. For windowed
(streaming) writes, each window's paths are emitted after that window is
finalized.
They also note the retry caveat raised in the issue thread: a
finalization retry doesn't re-emit files that an earlier attempt already
renamed, so the output may not list every file. I went ahead with option (1)
from my comment on #25769 (document the caveat; leave any behavior change for a
separate PR), since there was no reply after 5 business days.
`iobase.Write`'s docstring said it returned "a `pvalue.PValue` object
that represents the end of the Pipeline". It now describes the actual output,
and `Sink.finalize_write()` documents its return value.
2. **Fix (#41245).** In batch, `iobase.Write` emitted those paths at
`MAX_TIMESTAMP`, past the end of the global window. Grouping them downstream
failed the job on Prism, and `Count.Globally()` on them returned 0 on the
BundleBasedDirectRunner. They're now emitted at
`GlobalWindow().max_timestamp()`, as `_WriteBundleDoFn` has done since
BEAM-5250. `_WriteKeyedBundleDoFn`'s internal results get the same change for
consistency (they're only read as a side input).
The two parts are separate commits. I combined them because a test asserting
on the documented output can't pass on Prism without the fix. Happy to split
them into two PRs if you'd prefer.
**Testing** (Python 3.13, macOS, at a6ee50ed333 on top of 23012134b49)
- `pytest apache_beam/io/textio_test.py -k
'output_timestamps_in_global_window or outputs_final_file_paths'`: 3 passed.
These are the new tests: batch paths (`num_shards=3`), streaming paths
(`TestStream`, `triggering_frequency=10`), and batch output timestamps
(`num_shards` 0 and 3).
With `iobase.py` reverted to master, both batch tests fail on the default
test runner (Prism) and the streaming one still passes:
- `Failed assert: [Timestamp(9223371950454.774999)] ==
[Timestamp(9223372036854.775000)]`
- `assert_that` on the paths fails the job.
- `pytest -n 6
apache_beam/io/{textio,filebasedsink,iobase,avroio,parquetio,tfrecordio}_test.py`:
262 passed, 10 skipped.
- `pytest apache_beam/transforms/write_ptransform_test.py`: 4 passed.
- `pre-commit run --files <changed files>` (yapf, ruff): passed. `pylint
--rcfile=.pylintrc <changed .py files>`: 10.00/10.
- The repro from #41245 (`WriteToText(..., num_shards=3)` → `GroupByKey`,
and `Count.Globally()`), run on PrismRunner, BundleBasedDirectRunner and
FnApiRunner: all three now put the 3 paths in one group and count 3. On master,
Prism panics in watermark evaluation, the BundleBasedDirectRunner counts 0, and
FnApiRunner already works.
**Risk.** Only the timestamp of `Write`'s batch output and of its internal
keyed-writer results changes, from `MAX_TIMESTAMP` to the end of the global
window (about a day earlier). Windowed writes (`_finalize_windowed_write`)
aren't touched.
--
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]
