[BEAM-10670] Make Spark by default execute Read.Bounded using SplittableDoFn.

[lukecwik]

Instructions to use the old implementation are in CHANGES.md, latest version was merged in #13015 for release 2.25

---

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).
• Format the pull request title like [BEAM-XXX] Fixes bug in ApproximateQuantiles, where you replace BEAM-XXX with the appropriate JIRA issue, if applicable. This will automatically link the pull request to the issue.
• 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.

Post-Commit Tests Status (on master branch)

Lang	SDK	Dataflow	Flink	Samza	Spark	Twister2
Go		---		---		---
Java
Python				---		---
XLang		---		---		---

Pre-Commit Tests Status (on master branch)

---	Java	Python	Go	Website	Whitespace	Typescript
Non-portable
Portable	---		---	---	---	---

See .test-infra/jenkins/README for trigger phrase, status and link of all Jenkins jobs.

GitHub Actions Tests Status (on master branch)

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

[lukecwik]

R: @iemejia

Part 2 of #12603

Note that experimenting with watermark holds will likely take a few days to see if some of my ideas will work.

[lukecwik]

Run Spark ValidatesRunner

[lukecwik]

Run Spark StructuredStreaming ValidatesRunner

[iemejia]

Run Spark Runner Nexmark Tests

[lukecwik]

Run Spark ValidatesRunner

[lukecwik]

Run Spark StructuredStreaming ValidatesRunner

[lukecwik]

Run Spark Runner Nexmark Tests

[iemejia]

Run Spark Runner Nexmark Tests

[iemejia]

I am comparing the results of current master vs this PR in batch mode and the improvements are so big that I am even confused of how can it be so different, is it partitioning less or ignoring some operations because not need to estimate watermarks or something? Difference is really that good, Amazing!

Current master:

Performance:
Conf Runtime(sec) (Baseline) Events(/sec) (Baseline) Results (Baseline)
0000 2.1 47303.7 100000 0001 0.6 169779.3 92000 0002 0.3 293255.1 351 0003 3.1 32299.7 580 0004 1.0 10427.5 40 0005 1.5 67340.1 12 0006 1.1 9487.7 103 0007 1.7 59101.7 1 0008 1.3 77279.8 6000 0009 0.5 19084.0 298 0010 0.7 153139.4 1 0011 1.8 54112.6 1919 0012 0.9 112359.6 1919 0013 0.3 304878.0 92000 0014 0.9 113507.4 92000 ==========================================================================================

This PR

Performance:
Conf Runtime(sec) (Baseline) Events(/sec) (Baseline) Results (Baseline)
0000 1.1 90090.1 100000 0001 0.3 337837.8 92000 0002 0.1 694444.4 351 0003 1.4 71582.0 580 0004 1.0 10111.2 40 0005 0.6 177935.9 12 0006 0.3 40000.0 103 0007 0.4 227272.7 1 0008 0.3 314465.4 6000 0009 0.2 49019.6 298 0010 0.6 165016.5 1 0011 0.5 187969.9 1919 0012 0.2 492610.8 1919 0013 0.3 392156.9 92000 0014 0.9 113765.6 92000 ==========================================================================================

[lukecwik]

I am comparing the results of current master vs this PR in batch mode and the improvements are so big that I am even confused of how can it be so different, is it partitioning less or ignoring some operations because not need to estimate watermarks or something? Difference is really that good, Amazing!

Current master:

Performance:
Conf Runtime(sec) (Baseline) Events(/sec) (Baseline) Results (Baseline)
0000 2.1 47303.7 100000 0001 0.6 169779.3 92000 0002 0.3 293255.1 351 0003 3.1 32299.7 580 0004 1.0 10427.5 40 0005 1.5 67340.1 12 0006 1.1 9487.7 103 0007 1.7 59101.7 1 0008 1.3 77279.8 6000 0009 0.5 19084.0 298 0010 0.7 153139.4 1 0011 1.8 54112.6 1919 0012 0.9 112359.6 1919 0013 0.3 304878.0 92000 0014 0.9 113507.4 92000 ==========================================================================================

This PR

Performance:
Conf Runtime(sec) (Baseline) Events(/sec) (Baseline) Results (Baseline)
0000 1.1 90090.1 100000 0001 0.3 337837.8 92000 0002 0.1 694444.4 351 0003 1.4 71582.0 580 0004 1.0 10111.2 40 0005 0.6 177935.9 12 0006 0.3 40000.0 103 0007 0.4 227272.7 1 0008 0.3 314465.4 6000 0009 0.2 49019.6 298 0010 0.6 165016.5 1 0011 0.5 187969.9 1919 0012 0.2 492610.8 1919 0013 0.3 392156.9 92000 0014 0.9 113765.6 92000 ==========================================================================================

BoundedSource class doesn't expose a watermark API and the BoundedSourceAsSdfWrapper doesn't do much differently then the SourceRDD implementation. Partitioning is different, if you matched the partitioning between the SDF version and the non SDF version you'll see similar results. Flink saw something similar as well.

You can see that run conf 10 and 14 didn't have significantly different results.

[lukecwik]

@iemejia Merge?

[iemejia]

Yes definitely!

For the watermark part of my comment what I was expecting is that because we now need to deal with WatermarkEstimator + ProcessContinuation (even if dummy versions) for the SDF case it would maybe impact a bit the performance (+ the Restrictions moving around).

I am still unclear about why partitioning is better. From what I understood the Source is splitting in subsources that are the restrictions and from there we reshuffle, I expected similar results. So any extra detail would be nice to see what am I missing.

[iemejia]

LGTM Excellent improvement. Hopefully we can get rid of the native translations rapidly in the future. Thanks!

[lukecwik]

@iemejia How the number of partitions is calculated different during the SDF initial split then what we do with SourceRDD.

[lukecwik]

LGTM Excellent improvement. Hopefully we can get rid of the native translations rapidly in the future. Thanks!

We can remove it for each runner once they match parity with BoundedSource/UnboundedSource (done for all but Spark UnboundedSource) and after a few releases that have been issue free from users.

[iemejia]

@iemejia How the number of partitions is calculated different during the SDF initial split then what we do with SourceRDD.

If I understood correctly the initial split is only of size 1 because it is the full initial Source and then it divides further splitting based on the Source split, isn't it? I still expected that the reshuffle would distribute the work and we would end up with almost the same result after the subsequent split. Or did I misread it?