New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

[BEAM-8471] Flink native job submission for portable pipelines #9872

Merged

tweise merged 1 commit into apache:master from tweise:BEAM-8471.flinkPortableClientRunner

Oct 28, 2019

Contributor

tweise commented Oct 24, 2019 •

edited

Add a new Flink entry point (main method) that invokes the external SDK client entry point to generate the pipeline and submits the resulting Flink job like any other Flink native driver program would, via the optimizer plan environment ("[auto]").

Note that in this PR, the SDK client is assumed to be on the same host, which is the case when Flink and Python dependencies are in the same container image, for example. While this is something that can be solved at build time, the question from Robert made me realize that the dependency is almost identical to that between the runner and SDK on the execution side, which is abstracted via the environment concept. So we could, in the future, consider introducing a "client environment" or simply expand the existing environment. This would allow the SDK bits for both pipeline construction and execution to live in a side car container, separate from Flink.

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.
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	Apex	Dataflow	Flink	Gearpump	Samza	Spark
Go		---	---		---	---
Java
Python		---			---	---
XLang	---	---	---		---	---	---

Pre-Commit Tests Status (on master branch)

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

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

tweise mentioned this pull request

Entry point for Flink REST API using external SDK driver program. lyft/beam#27

Merged

tweise requested review from ibzib, mxm and robertwb

October 24, 2019 16:43

Contributor Author

tweise commented Oct 24, 2019

Contributor

chadrik commented Oct 24, 2019

Thanks for the heads up. I don't have any input on this, but I do like staying in the loop on this subject!

mxm reviewed

View reviewed changes

Contributor

mxm left a comment

Thanks @tweise. Looks good! A couple of minor comments and clarifications inline.

runners/flink/src/main/java/org/apache/beam/runners/flink/FlinkJobServerDriver.java Outdated Show resolved Hide resolved

runners/flink/src/main/java/org/apache/beam/runners/flink/FlinkPortableClientRunner.java Outdated Show resolved Hide resolved

runners/flink/src/main/java/org/apache/beam/runners/flink/FlinkPortableClientRunner.java Outdated Show resolved Hide resolved

runners/flink/src/main/java/org/apache/beam/runners/flink/FlinkPortableClientRunner.java Outdated

+                private void runDriverProgram() throws Exception {
+                  ProcessManager processManager = ProcessManager.create();
+                  String executable = "bash";

Contributor

mxm Oct 25, 2019

We may want to make this executable configurable, for non bash users. How about putting executable and args in the config with the current defaults?

runners/flink/src/main/java/org/apache/beam/runners/flink/FlinkPortableClientRunner.java Outdated Show resolved Hide resolved

runners/flink/src/main/java/org/apache/beam/runners/flink/FlinkPortableClientRunner.java Outdated Show resolved Hide resolved

runners/flink/src/main/java/org/apache/beam/runners/flink/FlinkPortableClientRunner.java Outdated Show resolved Hide resolved

runners/flink/src/main/java/org/apache/beam/runners/flink/FlinkPortableClientRunner.java Outdated Show resolved Hide resolved

runners/flink/src/main/java/org/apache/beam/runners/flink/FlinkPortableClientRunner.java Outdated

+                private int jobPort = 0; // any free port
+                public FlinkPortableClientRunner(String driverCmd) {
+                  this.driverCmd = driverCmd;

Contributor

mxm Oct 25, 2019

Suggested change

      
                this.driverCmd = driverCmd;
          
                Preconditions.checkState(!driverCmd.contains(DRIVER_CMD_FLAGS), "Driver command must not contain "+ DRIVER_CMD_FLAGS);
          
                this.driverCmd = driverCmd;

runners/flink/src/main/java/org/apache/beam/runners/flink/FlinkPortableClientRunner.java Outdated

+               */
+              public class FlinkPortableClientRunner {
+                private static final Logger LOG = LoggerFactory.getLogger(FlinkPortableClientRunner.class);
+                private static final String DRIVER_CMD_FLAGS = "--job_endpoint=%s";

Contributor

mxm Oct 25, 2019

Suggested change

      
              private static final String DRIVER_CMD_FLAGS = "--job_endpoint=%s";
          
              private static final String JOB_ENDPOINT_FLAG = "--job_endpoint";

Contributor

mxm Oct 25, 2019

This allows us to check for the args in the user command string. The %s would have to be appended in the format string.

mxm approved these changes

View reviewed changes

runners/flink/src/main/java/org/apache/beam/runners/flink/FlinkPortableClientEntryPoint.java Outdated Show resolved Hide resolved

tweise commented

View reviewed changes

runners/flink/src/main/java/org/apache/beam/runners/flink/FlinkPortableClientEntryPoint.java

+               *
+               * <p>Finally Flink launches the job.
+               */
+              public class FlinkPortableClientEntryPoint {

Contributor Author

tweise Oct 25, 2019

@mxm any preference regarding class name? I first had it as FlinkPortableClientRunner but maybe "runner" is misleading. On the other hand, "entry point" is also redundant.

Contributor

mxm Oct 28, 2019

How about just FlinkPortableRunner? The functionality is closest to the classic FlinkRunner, as it allows for any portable pipeline to be submitted.

Contributor Author

tweise Oct 28, 2019

I had it as XYZRunner originally, but figured it will cause confusion. There are FlinkPipelineRunner and PortablePipelineRunner already (these are used within the job server).

Contributor

mxm Oct 28, 2019

We could rename those 🤓

mxm reviewed

View reviewed changes

runners/flink/src/main/java/org/apache/beam/runners/flink/FlinkPortableClientEntryPoint.java

+               *
+               * <p>Finally Flink launches the job.
+               */
+              public class FlinkPortableClientEntryPoint {

Contributor

mxm Oct 28, 2019

How about just FlinkPortableRunner? The functionality is closest to the classic FlinkRunner, as it allows for any portable pipeline to be submitted.

runners/flink/src/main/java/org/apache/beam/runners/flink/FlinkPortableClientEntryPoint.java Outdated

+                  this.driverCmd = driverCmd;
+                }
+                /** Main method to be called within the Flink OptimizerPlanEnvironment. */

Contributor

mxm Oct 28, 2019

This comment is obsolete, no?

Contributor Author

tweise Oct 28, 2019

Yes, updated now.

tweise commented

View reviewed changes

...cution/src/test/java/org/apache/beam/runners/fnexecution/environment/ProcessManagerTest.java

+                @Ignore("not working yet")
+                @Test
+                public void testInheritIO() throws IOException, InterruptedException {

Contributor Author

tweise Oct 28, 2019 •

edited

@mxm I see the inherit working as the expected output is printed in the console. But short of starting a separate java process I don't see a way to capture it in the test.

Contributor Author

tweise Oct 28, 2019

I left the test but skipped the assertion, maybe we can still find a clever way to do this.


          [BEAM-8471] Flink native job submission for portable pipelines

f89c1b9

tweise force-pushed the BEAM-8471.flinkPortableClientRunner branch from 3b5dce8 to f89c1b9 Compare

October 28, 2019 20:02

tweise added a commit to lyft/beam that referenced this pull request


          [BEAM-8471] Entry point for Flink REST API using external SDK driver …

5ba66a6

…program. (#27)

Upstream PR: apache#9872

tweise merged commit 0645404 into apache:master

tweise deleted the BEAM-8471.flinkPortableClientRunner branch

October 28, 2019 22:27

tweise mentioned this pull request

[BEAM-8471] Put back FlinkJobServerDriver.fromParams to fix post commit #9911

Merged

3 tasks

mxm mentioned this pull request

[BEAM-8509] TestPortableRunner uses JobServerDriver interface #9914

Merged

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment