[BEAM-3287] Add Go support for universal runners, incl Flink #4888
Conversation
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
lostluck
requested changes
Mar 19, 2018
| @@ -28,12 +28,12 @@ import ( | |||
| const ( | |||
| // Model constants | |||
|
|
|||
| URNImpulse = "urn:beam:transform:impulse:v1" | |||
| URNImpulse = "beam:transform:impulse:v1" | |||
| URNParDo = "urn:beam:transform:pardo:v1" | |||
There was a problem hiding this comment.
Should ParDo still be prefixed with "urn:" ?
There was a problem hiding this comment.
Yes, that one is still the old way.
| // limitations under the License. | ||
|
|
||
| // Package runnerlib contains utilities for submitting Go pipelines | ||
| // to a Beam model runner. |
There was a problem hiding this comment.
Idle question & statement: Beam Model runner is what's we're going with for the name of portability runners long term? I like it.
There was a problem hiding this comment.
I don't think there is an agreed upon name long-term. The code package is "universal", so I'm not even internally consistent in this PR :)
| // Experiments are additional experiments. | ||
| Experiments []string | ||
|
|
||
| // TODO(herohde) 3/17/2018: add further parametrization as needed |
| } | ||
|
|
||
| // Prepare prepares a job to the given endpoint. It returns an id and endpoint, if successful. | ||
| func Prepare(ctx context.Context, client jobpb.JobServiceClient, p *pb.Pipeline, opt *JobOptions) (string, string, error) { |
There was a problem hiding this comment.
Since this function returns two parameters of the same type, it may be useful to use named parameters to be unambiguous about which returned value is which. eg (id, endpoint string, err error)
This is not advocating for naked returns being used.
| InternalJavaRunner string | ||
| } | ||
|
|
||
| // Prepare prepares a job to the given endpoint. It returns an id and endpoint, if successful. |
There was a problem hiding this comment.
We may be able to improve this comment, as I'm having difficulty reconciling what's here, and in the implementation.
Knowing nothing about what PrepareJob request on the service does, but reading this implementation, it looks like the idea is to prepare the runner to receive the job, largely by getting a place to stage artifacts (like the worker) to run the job later.
So, it's preparing the service for the job. Is that right?
With that in mind, and looking at how the returns are used in this PR perhaps:
// Prepare prepares a given Beam model runner to receive a job. If successful, Prepare returns an id, and a URL
// endpoint where the runner expects job artifacts to be staged.
There was a problem hiding this comment.
Yes. Rephrased the description.
| return resp.GetPreparationId(), resp.GetArtifactStagingEndpoint().GetUrl(), nil | ||
| } | ||
|
|
||
| // Submit submits a job to the given endpoint. It returns a jobID, if successful. |
There was a problem hiding this comment.
If we change the comment on Prepare, we should probably change "given endpoint" here to "given Beam model runner" for consistency.
There was a problem hiding this comment.
Made them consistent.
| "github.com/apache/beam/sdks/go/pkg/beam/util/grpcx" | ||
| ) | ||
|
|
||
| // Stage stages the worker binary and any additional content to the given endpoint. |
There was a problem hiding this comment.
Should "content" be changed to "artifacts" for consistency?
There was a problem hiding this comment.
Change it to files, which is what this helper function accepts.
|
I think I'm not really helpful since I don't really know Go but the structure of this looks good! Did you actually try this against the hacking job server we have? What did you get? |
* Extracted a runnerlib for runners that wish to use the portable model, but adds special logic or integration.
|
@aljoscha Thanks! Yes! That is how I'm testing this stuff. I'm currently hitting an issue with the artifacts on the Flink side: Flink runner logs (info): |
|
@lostluck Thanks! PTAL |
|
Thanks @aljoscha. I figured it was something simple. |
|
Now I get: It seems we're mangling the name. The correct file, named "worker", is indeed in this directory: |
|
Yes, you're right. This was a case of misapplied mangling. It should be fixed now: |
|
Tried to remove the demangling and using the go container and wordcount seems to work: |
|
@axelmagn Tested with your changes and it seems to work as well! Awesome! |
lostluck
approved these changes
Mar 20, 2018
| beam.RegisterRunner("universal", Execute) | ||
| } | ||
|
|
||
| // Execute execute the pipeline on a universal beam runner. |
lostluck
approved these changes
Mar 20, 2018
| beam.RegisterRunner("universal", Execute) | ||
| } | ||
|
|
||
| // Execute execute the pipeline on a universal beam runner. |
lostluck
approved these changes
Mar 20, 2018
| beam.RegisterRunner("universal", Execute) | ||
| } | ||
|
|
||
| // Execute execute the pipeline on a universal beam runner. |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
Extracted a runnerlib for runners that wish to use the portable model, but adds special logic or integration. Updated the portable URNs.
To run against a Flink runner, do from sdks/go:
$ go run examples/wordcount/wordcount.go --runner=flink --output=/tmp/foo --endpoint=localhost:3000