[
https://issues.apache.org/jira/browse/BEAM-5724?focusedWorklogId=158915&page=com.atlassian.jira.plugin.system.issuetabpanels:worklog-tabpanel#worklog-158915
]
ASF GitHub Bot logged work on BEAM-5724:
----------------------------------------
Author: ASF GitHub Bot
Created on: 25/Oct/18 22:11
Start Date: 25/Oct/18 22:11
Worklog Time Spent: 10m
Work Description: mwylde opened a new pull request #6835: [BEAM-5724]
Generalize flink executable context to allow more than 1 worker process per
task manager
URL: https://github.com/apache/beam/pull/6835
By default, beam will create a single python sdk worker per Flink
taskmanager / job, and #6524 added an option (`--sdk-worker-parallelism stage`)
that would instead create one python worker per Flink stage. However in our
experience this leads to far too many python processes (possibly thousands), as
flink may run many stages in a single task slot.
This PR introduces an alternate approach to managing sdk worker concurrency.
Basically, we know that our box has some capacity (roughly correlated to number
of CPU cores) and we'd like to use as much as possible for executing our worker
code, while not suffering the overhead of too many worker processes.
With this PR, `--sdk-worker-parallelism` now takes the number of processes
to run per TM. By default this is 1 (retaining the existing behavior). There's
also an "auto" value of 0 that will set it to `number of cores - 1`, which is
likely what most streaming users would want. As stages are constructed, we will
create new StageContexts until we have `--sdk-worker-parallelism` of them, then
we round-robin assign new stages to the existing contexts.
cc @tweise
------------------------
Follow this checklist to help us incorporate your contribution quickly and
easily:
- [x] 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.
- [ ] If this contribution is large, please file an Apache [Individual
Contributor License Agreement](https://www.apache.org/licenses/icla.pdf).
It will help us expedite review of your Pull Request if you tag someone
(e.g. `@username`) to look at it.
Post-Commit Tests Status (on master branch)
------------------------------------------------------------------------------------------------
Lang | SDK | Apex | Dataflow | Flink | Gearpump | Samza | Spark
--- | --- | --- | --- | --- | --- | --- | ---
Go | [](https://builds.apache.org/job/beam_PostCommit_Go_GradleBuild/lastCompletedBuild/)
| --- | --- | --- | --- | --- | ---
Java | [](https://builds.apache.org/job/beam_PostCommit_Java_GradleBuild/lastCompletedBuild/)
| [](https://builds.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Apex_Gradle/lastCompletedBuild/)
| [](https://builds.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Dataflow_Gradle/lastCompletedBuild/)
| [](https://builds.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Flink_Gradle/lastCompletedBuild/)
[](https://builds.apache.org/job/beam_PostCommit_Java_PVR_Flink/lastCompletedBuild/)
| [](https://builds.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Gearpump_Gradle/lastCompletedBuild/)
| [](https://builds.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Samza_Gradle/lastCompletedBuild/)
| [](https://builds.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Spark_Gradle/lastCompletedBuild/)
Python | [](https://builds.apache.org/job/beam_PostCommit_Python_Verify/lastCompletedBuild/)
| --- | [](https://builds.apache.org/job/beam_PostCommit_Py_VR_Dataflow/lastCompletedBuild/)
</br> [](https://builds.apache.org/job/beam_PostCommit_Py_ValCont/lastCompletedBuild/)
| [](https://builds.apache.org/job/beam_PostCommit_Python_VR_Flink/lastCompletedBuild/)
| --- | --- | ---
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
[email protected]
Issue Time Tracking
-------------------
Worklog Id: (was: 158915)
Time Spent: 10m
Remaining Estimate: 0h
> Beam creates too many sdk_worker processes with --sdk-worker-parallelism=stage
> ------------------------------------------------------------------------------
>
> Key: BEAM-5724
> URL: https://issues.apache.org/jira/browse/BEAM-5724
> Project: Beam
> Issue Type: Improvement
> Components: runner-flink
> Reporter: Micah Wylde
> Assignee: Micah Wylde
> Priority: Major
> Labels: portability-flink
> Time Spent: 10m
> Remaining Estimate: 0h
>
> In the flink portable runner, we currently support two options for sdk worker
> parallelism (how many python worker processes we run). The default is one per
> taskmanager, and with --sdk-worker-parallelism=stage you get one per stage.
> However, for complex pipelines with many beam operators that get fused into a
> single flink task this can produce hundreds of worker processes per TM.
> Flink uses the notion of task slots to limit resource utilization on a box; I
> think that beam should try to respect those limits as well. I think ideally
> we'd produce a single python worker per task slot/flink operator chain.
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
