Python TextIO Performance Test

[Abacn]

Part of #18204

Creates Python TextIO Performance Test. The jenkins task mirrors its Java counterpart and more tests (AvroIO, TFRecordsIO, etc) will be added in followup PRs.

Please add a meaningful description for your change here

---

Thank you for your contribution! Follow this checklist to help us incorporate your contribution quickly and easily:

• Choose reviewer(s) and mention them in a comment (R: @username).
• Mention the appropriate issue in your description (for example: addresses #123), if applicable. This will automatically add a link to the pull request in the issue. If you would like the issue to automatically close on merging the pull request, comment fixes #<ISSUE NUMBER> instead.
• Update CHANGES.md with noteworthy changes.
• If this contribution is large, please file an Apache Individual Contributor License Agreement.

See the Contributor Guide for more tips on how to make review process smoother.

To check the build health, please visit https://github.com/apache/beam/blob/master/.test-infra/BUILD_STATUS.md

GitHub Actions Tests Status (on master branch)

See CI.md for more information about GitHub Actions CI.

[codecov]

Codecov Report

Merging #23951 (31b469b) into master (e2ea0c7) will decrease coverage by 0.07%.
The diff coverage is 71.42%.

@@            Coverage Diff             @@
##           master   #23951      +/-   ##
==========================================
- Coverage   73.52%   73.45%   -0.08%     
==========================================
  Files         707      709       +2     
  Lines       95856    95985     +129     
==========================================
+ Hits        70481    70502      +21     
- Misses      24058    24166     +108     
  Partials     1317     1317              

Flag	Coverage Δ
python	83.18% <71.42%> (-0.13%)	⬇️

Flags with carried forward coverage won't be shown. Click here to find out more.

Impacted Files	Coverage Δ
...s/python/apache_beam/testing/synthetic_pipeline.py	82.29% <ø> (ø)
...beam/testing/load_tests/load_test_metrics_utils.py	33.82% <71.42%> (-0.26%)	⬇️
sdks/python/apache_beam/utils/interactive_utils.py	95.12% <0.00%> (-2.44%)	⬇️
.../python/apache_beam/transforms/periodicsequence.py	98.50% <0.00%> (-1.50%)	⬇️
...hon/apache_beam/runners/worker/bundle_processor.py	93.17% <0.00%> (-0.25%)	⬇️
...ks/python/apache_beam/runners/worker/sdk_worker.py	89.08% <0.00%> (-0.17%)	⬇️
...ference/tfx_bsl/tensorflow_image_classification.py	0.00% <0.00%> (ø)
...amples/inference/tfx_bsl/build_tensorflow_model.py	0.00% <0.00%> (ø)
sdks/python/apache_beam/transforms/core.py	92.90% <0.00%> (+0.03%)	⬆️
...s/python/apache_beam/io/gcp/bigquery_file_loads.py	87.41% <0.00%> (+0.08%)	⬆️
... and 3 more

📣 We’re building smart automated test selection to slash your CI/CD build times. Learn more

[github-actions]

Assigning reviewers. If you would like to opt out of this review, comment assign to next reviewer:

R: @ryanthompson591 for label python.
R: @johnjcasey for label io.

Available commands:

• stop reviewer notifications - opt out of the automated review tooling
• remind me after tests pass - tag the comment author after tests pass
• waiting on author - shift the attention set back to the author (any comment or push by the author will return the attention set to the reviewers)

The PR bot will only process comments in the main thread (not review comments).

[Abacn]

stop reviewer notifications

[github-actions]

Stopping reviewer notifications for this pull request: requested by reviewer

[Abacn]

Runs performance tests for Python TextIOIT

[Abacn]

Run Python TextIO Performance Test

[Abacn]

Run Python TextIO Performance Test

[Abacn]

Run Python TextIO Performance Test

[Abacn]

Run Python TextIO Performance Test

[Abacn]

Run Python TextIO Performance Test

[Abacn]

Run Python TextIO Performance Test

[Abacn]

Run Python TextIO Performance Test

[Abacn]

Run Python TextIO Performance Test

[Abacn]

Test passed: https://ci-beam.apache.org/job/beam_PerformanceTests_TextIOIT_Python/8/
with metrics:
INFO:apache_beam.testing.load_tests.load_test_metrics_utils:Metric: write_time_runtime Value: 604
INFO:apache_beam.testing.load_tests.load_test_metrics_utils:Metric: read_time_runtime Value: 356
cleaning up

[Abacn]

Run Python TextIO Performance Test

[Abacn]

Test passed: https://ci-beam.apache.org/job/beam_PerformanceTests_TextIOIT_Python/9/

[Abacn]

Ready for review now. R: @johnjcasey

[ryanthompson591]

R: @AnandInguva Adding Anand as he is really familiar with groovy files.

[ryanthompson591]

what does H H mean?

[Abacn]

H 15 * * *' means the job will initiate every day 15:XX PM and currently all Python IO performance test is running on this schedule. As more tests being added we should distribute the time evenly throughout a day, which is 'H H * * *' --- both minute and hour are random

[ryanthompson591]

On line 108 you added a batch test, does that mean this one is streaming?

[Abacn]

Same change as above, randomize cron job time.

[ryanthompson591]

Feel free to add another reviewer for these graphana changes as I'm not really familiar enough to give good feedback.

[Abacn]

Thanks. This trivially adds a graph, see #23914 for some context.

[ryanthompson591]

IMO, things that you will not configure can be hard coded. Will dataset_size always be the same?

I'll leave keeping this or removing up to you.

[Abacn]

FileBasedIOTestOptions mirrors its java counterpart: https://github.com/apache/beam/blob/master/sdks/java/io/file-based-io-tests/src/test/java/org/apache/beam/sdk/io/common/FileBasedIOTestPipelineOptions.java

As far as I understood dataset_size is for documentation purpose and is added as an entry of the metric in Java. I preserved this argument in alignment with Java implementation, though currently it is not used elsewhere. It is now purely documentation purpose that provides an expected dataset size in the test options.

For uncompressed, the actual dataset size is always the same; for compressed text, it may fluctuate.

[ryanthompson591]

this output length is a little hard to read. Is this machine read somewhere? If not maybe just describe it?

[Abacn]

It is just a copy-paste of Base64 length calculation: https://stackoverflow.com/questions/13378815/base64-length-calculation. Will add more informative comments here

[ryanthompson591]

AFAICT this test is only implemented in one place and need_cleanup is always false, why not just remove this variable?

[Abacn]

I try to implement each class in a way that it is still a complete load test (doing setup and cleanup) if invoked alone, as it is inherit from LoadTest. The test suite runs WriteTest and ReadTest sequentially to avoid setup the resources twice.

[ryanthompson591]

I'm worried about a race case here. Since the folder creation is parallel to the writing, I'm worried it could happen that the folder creation could happen too late.

What do you think of this change.

after_folder_creation = (
        self.pipeline
        | beam.Impulse()
        | beam.ParDo(CreateFolderFn(self.output_folder)))

  # write to text
    _ = (
        after_folder_creation ### Make sure this happens after the creation of a folder.
        | 'Produce rows' >> Read(
            SyntheticSource(self.parse_synthetic_source_options()))

[Abacn]

I have to create the folder at pipeline runtime because the tests should work fine when only pipeline runner has access to the filesystem, and the submitter may not have.
For GCS filesystem it is fine. mkdirs is a non-op as GCS is a flat structure. Here it is aimed to make the test run in both nested filesystem (like local filesystem) and flat filesystem (like gcs).
The change would be the same because the Read transform essentially expands from pbegin (it also uses Impulse which creates a PCollection from pbegin.pipeline). The pipeline graph will remain the same. Tried that Read() must follow pbegin because it uses Impulse() which asserts the upstream is pbegin.

[ryanthompson591]

Why do we need Impulse here?

[Abacn]

Just a common pattern followed.

[ryanthompson591]

Why do we need to format these records?

[Abacn]

The bytes emitted from SyntheticSource are Key Value pairs with random bytes. We need to convert it to a text line (ideally a line of ascii chars and does not contain line break \n)

[ryanthompson591]

what does this base class gain for us?

[Abacn]

work with type hint here

beam/sdks/python/apache_beam/testing/load_tests/load_test_metrics_utils.py

Line 206 in 560aa8d

self.publishers: List[MetricsPublisher] = []

as publish method is invoked here

beam/sdks/python/apache_beam/testing/load_tests/load_test_metrics_utils.py

Line 236 in 560aa8d

publisher.publish(insert_dicts)

[ryanthompson591]

LGTM

[Abacn]

Python precommit failed test known flakes: #22115

R: @johnjcasey for final review, thanks!