[flink-runner] Improve Datastream for batch performances #32440

jto · 2024-09-12T09:18:52Z

Context

Flink will drop support for the dataset API in 2.0 which should be released by EOY so it quite important for Beam to support Datastream well.

The PR

This PR improves the performances of Batch jobs executed with --useDatastreamForBatch by porting the following performance optimizations already present in FlinkBatchTransformTranslators but lacking in FlinkStreamingTransformTranslators.

Limit the max size of source splits. Similar to Reduce the maximum size of input splits in Flink to better distribute work #28045
Pre-combine before shuffle (both reduce by key and GBK)
Disable bundling in batch mode (except for pre-combine). Lower the default bundle size since the new behavior puts pressure on the heap.

It also implements the following optimizations:

Use a "lazy" split enumerator to distributes split dynamically rather than eagerly. This new enumerator greatly reduces skew as each slot is able to pull new splits to consume only when it has finished its work.
Set the default maxParallelism to parallelism as the total number of key groups is equal to maxParallelism. Again this reduces skew.
Make ToKeyedWorkItem part of DoFnOperator which reduces the size of the job graph and avoid unnecessary inter-task communication.
Force a common slot-sharing group on every bounded IOs. This emulate the behavior of the Dataset API which again improves performances especially when data is being shuffled several times while partitioning keys are unchanged (for example of the job does GBK -> map -> CombinePerKey). Add a flag to control this feature (defaults to active).
Use a custom class for keys which guarantee a good distribution of data even when the number of keys is not >> parallelism.
Other minor optimizations removing repeated serde work.

Benchmarks

The patched version was tested against a few of Spotify's production batch workflows. All settings were left unchanged except for the followings:

passed --useDatastreamForBatch=true
set jobmanager.scheduler: default (otherwise datastream default to adaptive scheduler).

		Beam 2.56 - dataset	Beam 2.56 - datastream		Beam 2.56 - datastream patched
job	# workers	execution time	execution time	% diff	execution time	% diff
Job 1	350	2:19:00	fails after 4h29min	-	1:43:00	-25.90%
Job 2	160	0:23:00	0:35:00	52.17%	0:22:36	-1.74%
Job 3	200	0:53:08	1:34:39	78.14%	failed	-
Job 4	160	2:31:20	4:27:00	76.43%	2:19:35	-7.76%
Job 5	1	0:43:00	not tested	-	0:38:00	-11.63%
Job 6	300	3:01:00	not tested	-	2:55:00

Note

Job 3 fails with a stackoverflow exception because of a bug in versions of Kryo < 3.0. I believe this is because the job uses taskmanager.runtime.large-record-handler: true and it should be fixed in Flink 2.0 since Kryo is upgraded to a more recent version.

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

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 or the workflows README to see a list of phrases to trigger workflows.

github-actions · 2024-09-13T10:37:25Z

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

R: @damccorm for label website.

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).

damccorm · 2024-09-13T17:18:22Z

@kennknowles would you mind taking a look at this one?

github-actions · 2024-09-24T12:14:11Z

Reminder, please take a look at this pr: @damccorm

github-actions · 2024-09-26T12:14:19Z

Assigning new set of reviewers because Pr has gone too long without review. If you would like to opt out of this review, comment assign to next reviewer:

R: @melap for label website.

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)

damccorm · 2024-10-03T15:34:02Z

R: @kennknowles

github-actions · 2024-10-03T15:43:06Z

Stopping reviewer notifications for this pull request: review requested by someone other than the bot, ceding control. If you'd like to restart, comment assign set of reviewers

kennknowles · 2024-10-03T20:39:05Z

To test this thoroughly, let us add some of the postcommits by touching trigger files. In #32648 you can see how I edited the JSON files (including some new ones) and I think these are all the Flink-specific postcommit jobs.

jto · 2024-10-17T15:37:14Z

@kennknowles done. I also rebased master but some of the tests seem to be quite flaky now. There are test failing on things I did not touch (direct runner) and the Flink tests that are failing here are not failing on my machine... Any idea how I could make them work ?

kennknowles · 2024-10-18T17:22:17Z

I opened jto#236 with some more trigger files. The "PVR" trigger files stands for "Portable Validates Runner" that isn't as directly impacted. I think the non-portable ValidatesRunner tests should test that the runner still complies with the model and passes the basic tests.

jto · 2024-11-05T09:54:33Z

Thanks! I just merged it.
Sorry for the slow response. I was on vacation :)

jto · 2024-11-18T13:19:48Z

Hey @kennknowles!
I tried really hard to make the failing test (org.apache.beam.sdk.transforms.ParDoTest$LifecycleTests) work but in the end I think it's impossible.
The behaviour of the runner is correct (it fails when it's supposed to) but Flink will either fail with the expected exception or just fail with a generic error (TaskNotRunningException: Task is not running, but in state FAILED).
The expected error is there in the logs but the "final" error may or may not be the expected one. The behaviour is non-deterministic.
In the end I just added this class into sickbay and that seems to have fixed almost all the problems.

The Python PostCommits are failing but the error is:

2024-11-18T11:39:33.3464029Z Execution failed for task ':sdks:python:container:py312:docker'.                                                                                                              │
│2024-11-18T11:39:33.3464463Z > Process 'command 'docker'' finished with non-zero exit value 1

which I think is unrelated to those changes and I could not find anything in the logs suggesting otherwise.

claudevdm · 2025-01-07T14:22:12Z

Hey @kennknowles @jto does this PR have any next steps?

kennknowles · 2025-01-10T15:40:10Z

I agree with your change to sickbay that Lifecycle test. It sounds like it is incorrectly specified, not taking into account allowable runner variation. Sorry for the delay which has led to conflicts - if you resolve them I think this is good to merge.

jto · 2025-01-27T09:13:21Z

Hey @kennknowles !

Thanks for following up. I merged the latest changes from master and the PR can now be merged.

I had to add another test to the sickbay. org.apache.beam.sdk.transforms.ViewTest.testTriggeredLatestSingleton makes both validatesRunnerBatch and validatesRunnerBatchWithDataStream fail with a timeout. Technically the test passes in both cases but for some reason it takes a very long time to execute. Because it causes the same issue with and without DataStream, I don't think this PR is the cause.

Some of the Python tests are failing with the following exception:

ValueError: numpy.dtype size changed, may indicate binary incompatibility. Expected 96 from C header, got 88 from PyObject

which does not look like to be caused by this PR.

:sdks:java:harness:test failed after ~6h which again looks unrelated to this PR.

It'd be ideal if fixes were available and could be rebased into my branch to get all the tests green before merging but otherwise I think the changes in the Flink runner are in a good shape.

jto · 2025-01-27T09:17:24Z

Also history is a bit messy because I tested a bunch of options to get the test to pass. Cleaning it up right now :)

…stSingleton to sickbay restore gradle parallel mode

jto · 2025-01-27T13:29:13Z

Ready for merge :)

kennknowles · 2025-02-03T14:24:22Z

runners/flink/flink_runner.gradle

@@ -251,7 +257,7 @@ def createValidatesRunnerTask(Map m) {
    def pipelineOptionsArray = ["--runner=TestFlinkRunner",
                                "--streaming=${config.streaming}",
                                "--useDataStreamForBatch=${config.useDataStreamForBatch}",
-                                "--parallelism=2",
+                                "--parallelism=1",


This seems like it could be a problem, actually. Why did you have to do this? Memory pressure on test workers?

Yes I think validate runner was getting OOMKilled

kennknowles · 2025-02-03T14:25:19Z

runners/flink/src/test/validatesRunnerConfig/flink-conf.yaml

+#parallelism.default: 23
+taskmanager.memory.network.fraction: 0.2
+taskmanager.memory.network.max: 1gb
+pipeline.operator-chaining.enabled: false


Is this OK? I guess it forces more serialization so everything will get slower during testing, but coder errors will be caught. Could it also mask errors?

Hum tbh I don't really remember why I changed that. I'll check maybe this is not required.

Hey! So at least locally it does not break the test. I'll make another PR to revert this change.

jto · 2025-02-03T15:41:30Z

Thanks for reviewing and merging :)

jto · 2025-02-03T16:13:21Z

@kennknowles made a PR here: #33838

* [Flink] Set return type of bounded sources * [Flink] Use a lazy split enumerator for bounded sources * [Flink] fix lazy enumerator package * [Flink] Default to maxParallelism = parallelism in batch * [Flink] Avoid re-serializing trigger on every element * [Flink] Avoid re-evaluating options every time a new state is stored * [Flink] Only serialize states namespace keys if necessary * [Flink] Make ToKeyedWorkItem part of the DoFnOperator * [Flink] Remove ToBinaryKV * [Flink] Refactor CombinePerKeyTranslator * [Flink] Combine before Reduce (no side-input only) * [Flink] Implement partial reduce * [Flink] dead code cleanup * [Flink] persistent PartialReduceBundleOperator operator state * [Flink] Combine before GBK * [Flink] Combine before reduce (with side input) * [Flink] Force slot sharing group in batch mode * [Flink] Disable bundling in batch mode * [Flink] Lower default max bundle size in batch mode * [Flink] Code cleanup * [Flink] fix WindowDoFnOperatorTest * [Flink] Remove 1.14 compat code * [Flink] Fix flaky test * [Flink] Use a custom key type to better distribute load * [Flink] fix GBK streaming with side input * [Flink] fix error management in lazy source * [Flink] disable operator chaining in validatesRunner * [Flink] fix lazy source enumerator behaviour on error * [Flink] set validates runner parallelism to 1 * [Flink] add org.apache.beam.sdk.transforms.ParDoTest to sickbay * [Flink] fix lazy source enumerator behaviour on error (again) * [Flink] Add org.apache.beam.sdk.transforms.ViewTest.testTriggeredLatestSingleton to sickbay

github-actions bot added website runners core flink labels Sep 12, 2024

jto changed the title ~~Julient/batch datastream~~ Improve Datastream for batch performances Sep 12, 2024

jto force-pushed the julient/batch-datastream branch 3 times, most recently from fea7323 to 893c19f Compare September 13, 2024 09:31

jto changed the title ~~Improve Datastream for batch performances~~ [flink-runner] Improve Datastream for batch performances Sep 13, 2024

jto marked this pull request as ready for review September 13, 2024 10:24

github-actions bot added the Next Action: Reviewers label Sep 13, 2024

github-actions bot added the slow-review label Sep 24, 2024

github-actions bot removed the slow-review label Sep 26, 2024

kennknowles self-requested a review September 26, 2024 16:44

jto force-pushed the julient/batch-datastream branch from d88deed to e2fc26e Compare October 16, 2024 18:52

github-actions bot added the build label Oct 16, 2024

jto force-pushed the julient/batch-datastream branch 2 times, most recently from e060213 to ae0f3b4 Compare October 17, 2024 14:00

kennknowles mentioned this pull request Oct 18, 2024

Additional Flink github action trigger files jto/beam#236

Merged

4 tasks

github-actions bot added the java label Nov 6, 2024

jto added 2 commits November 14, 2024 10:12

[Flink] set validates runner parallelism to 1

777dcf7

[Flink] add org.apache.beam.sdk.transforms.ParDoTest to sickbay

6161f4e

jto added 3 commits January 22, 2025 15:39

[Flink] fix lazy source enumerator behaviour on error (again)

7a766f3

[Flink] merge master

e8e00d4

[Flink] spotless

f6b8779

github-actions bot added java io aws labels Jan 22, 2025

jto added 3 commits January 23, 2025 13:59

Merge remote-tracking branch 'origin' into julient/batch-datastream

10c480c

[Flink] merge

cb9c6b2

Disable gradle parallel execution

dd23ef0

jto force-pushed the julient/batch-datastream branch 2 times, most recently from cfdfbfd to 7f83bbe Compare January 24, 2025 09:32

[Flink] merge again

24a5a51

jto force-pushed the julient/batch-datastream branch from 7f83bbe to 24a5a51 Compare January 24, 2025 09:46

[Flink] Add org.apache.beam.sdk.transforms.ViewTest.testTriggeredLate…

aaaa023

…stSingleton to sickbay restore gradle parallel mode

jto force-pushed the julient/batch-datastream branch from e147b02 to aaaa023 Compare January 27, 2025 10:15

kennknowles approved these changes Feb 3, 2025

View reviewed changes

kennknowles merged commit 83b3735 into apache:master Feb 3, 2025
35 of 39 checks passed

kennknowles reviewed Feb 3, 2025

View reviewed changes

jto mentioned this pull request Feb 3, 2025

Revert "[Flink] disable operator chaining in validatesRunner" #33838

Merged

3 tasks

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

[flink-runner] Improve Datastream for batch performances #32440

[flink-runner] Improve Datastream for batch performances #32440

jto commented Sep 12, 2024 •

edited

Loading

github-actions bot commented Sep 13, 2024

damccorm commented Sep 13, 2024

github-actions bot commented Sep 24, 2024

github-actions bot commented Sep 26, 2024

damccorm commented Oct 3, 2024

github-actions bot commented Oct 3, 2024

kennknowles commented Oct 3, 2024

jto commented Oct 17, 2024 •

edited

Loading

kennknowles commented Oct 18, 2024

jto commented Nov 5, 2024

jto commented Nov 18, 2024

claudevdm commented Jan 7, 2025

kennknowles commented Jan 10, 2025

jto commented Jan 27, 2025

jto commented Jan 27, 2025 •

edited

Loading

jto commented Jan 27, 2025

kennknowles Feb 3, 2025

jto Feb 3, 2025

kennknowles Feb 3, 2025

jto Feb 3, 2025

jto Feb 3, 2025

jto commented Feb 3, 2025

jto commented Feb 3, 2025

[flink-runner] Improve Datastream for batch performances #32440

[flink-runner] Improve Datastream for batch performances #32440

Conversation

jto commented Sep 12, 2024 • edited Loading

Context

The PR

Benchmarks

Note

GitHub Actions Tests Status (on master branch)

github-actions bot commented Sep 13, 2024

damccorm commented Sep 13, 2024

github-actions bot commented Sep 24, 2024

github-actions bot commented Sep 26, 2024

damccorm commented Oct 3, 2024

github-actions bot commented Oct 3, 2024

kennknowles commented Oct 3, 2024

jto commented Oct 17, 2024 • edited Loading

kennknowles commented Oct 18, 2024

jto commented Nov 5, 2024

jto commented Nov 18, 2024

claudevdm commented Jan 7, 2025

kennknowles commented Jan 10, 2025

jto commented Jan 27, 2025

jto commented Jan 27, 2025 • edited Loading

jto commented Jan 27, 2025

kennknowles Feb 3, 2025

Choose a reason for hiding this comment

jto Feb 3, 2025

Choose a reason for hiding this comment

kennknowles Feb 3, 2025

Choose a reason for hiding this comment

jto Feb 3, 2025

Choose a reason for hiding this comment

jto Feb 3, 2025

Choose a reason for hiding this comment

jto commented Feb 3, 2025

jto commented Feb 3, 2025

jto commented Sep 12, 2024 •

edited

Loading

jto commented Oct 17, 2024 •

edited

Loading

jto commented Jan 27, 2025 •

edited

Loading