Cross-language Beam Pipelines
Thomas Groh (tgroh at google dot com)
Allow pipeline authors to construct Pipelines containing transforms authored in multiple languages within Beam SDKs.
The Beam Fn API supports executing code written in arbitrary languages, but there is no officially supported way for a pipeline author to write a pipeline which uses code written in multiple languages. Enabling transforms to be authored in one SDK and invoked elsewhere enables reuse of transforms across SDKs without the transform author having to reimplement the entire transform, even so far as to not implementing an SDK beyond the ability to construct a runner API graph. This reduces the burden on new SDK or connector authors to reach a minimum level of functionality for all SDKs past the first. This enables SDKs to share connectors and other complex transforms, as well as transforms that utilize language-specific constructs and libraries. Users, meanwhile, can build a pipeline in the language of their choice using the appropriate idioms of that language.
There are two different concerns that must be answered when determining how cross-language pipelines are expressed and executed[e]:
Most methods to expand a cross-language PTransform can be performed in both the contexts of an SDK or a pipeline runner. However, there are environmental and updatability concerns depending on the environment which differ depending on how cross-language PTransform are expanded, which will be discussed in each expansion method.
Designs of how cross-language transforms share configuration is a separate concern that will not be considered in this document, beyond the expectation that the FunctionSpec used by the [f]Runner API is suitable for transmitting an URN (to identify the UDF) and the configuration (via an opaque blob). More specifically, this document will be written with the assumption that some registration scheme exists to enable arbitrary transforms to be constructed from a configuration within an SDK. There are additional restrictions on which transforms can be used in a cross-language manner; not least among which is that the Coder of all inputs and all outputs must be understood in both languages.
Within this document, the SDK which is capable of expanding a transform will be referred to as the “target” environment.
Note that the following proposals are not mutually exclusive. They define different ways in which a cross language pipeline can be expanded, each with different tradeoffs, and it is possible that all of the proposals are implemented and used within the Beam SDKs.
Each language SDK provides a harness which exposes an RPC API, approximately:
rpc Expand (ExpandRequest) returns (ExpandResponse) message ExpandRequest { PTransform transform = 1; Components components = 2; } message ExpandResponse { PTransform expanded = 1; Components components = 2; } |
To produce the final transform, the transform in the original SDK produces a shim node, which produces a collection of ‘primitive’ outputs. To expand this shim, this node, plus all input and output PCollections and associated components (Coders, WindowingStrategies, etc) as the contents of the ExpandRequest to the harness provided by the target environment. Within the target environment, the SDK plus any required extensions construct the composite transform from the payload of the transform contained within the request, and returns the expansion in the ExpandResponse.
The choice of specific process is mostly not considered here - examples include a per-SDK native process (invoked by directly invoking an appropriate program written in the same language as the target SDK), or a Docker container, or an externally hosted service.
Each SDK will require logic to perform routing of the configuration stub to an appropriate implementation, or return an error if none is available.
To invoke a cross-language transform, an SDK inserts all transform nodes that would be generated by invoking the same transform in the target language. Each primitive node includes a reference to an environment in which the target UDF can be constructed, and configuration for that UDF.
During execution time, the harness for that environment initializes the target transform from the configuration contained within the primitive node.
Include within the Beam SDKs a module which contains static expansions, expressed in the form of a Runner API subgraph and consumed by all SDKs. When a cross-language transform is encountered, insert the expansion that corresponds to the urn and payload of the transform.
The process to produce a precompiled expanded form is not yet developed.
[1]Because it decouples the means of expansions from their contents. With an appropriate environment, arbitrary transforms can be expanded. This could even include non-SDK expansions when appropriate
[2] For critical bug fixes, performance updates, etc
[a]Do you have a concrete (sketch of) example for how it would look from the user's point of view? I.e., what would they actually write to invoke a TfTransform from Java, say?
[b]+tgroh@google.com I was also looking for an example of how the user would express a cross-language pipeline, is that covered elsewhere?
[c]The invocation should be transparent to the user (e.g. if I invoke a python transform from Java, I do so as though it was a native Java transform) - there are difficulties for more complex transforms, such as running code in a separate process/within a docker container - however, that's a concern about the environment rather than the user-facing API
[d]For the library author of the cross-language transform, that's an undesigned API, with two different components.
For topologically simple transforms, SDKs introduce a method by which a transform can be instantiated within an SDK harness via a payload (much like known Coders can be deserialized via URN and components), and a way to populate appropriate configuration within the host SDKs, including an environment capable of executing the transform.
For topologically-relevant transforms we introduce some RPC layer, and SDKs provide wrappers around it to identify the remote environment, which constructs that environment and sends an RPC for it to construct a subgraph.
[e]There is also the question of how the runtime execution requirements are satisfied, notably staged files for the shared transform. We may need to create tooling to bundle everything into a container image for shared (cross-language) transforms.
[f]Also many transforms are configured with code (e.g. SerializableFunctions). We either need a way to manage that, or create simpler (possibly more-restricted) wrappers around these transforms that can be exposed cross language.
[g]This is interesting. What would be the benefit/use of such a service?
[h]Presumably we can do this even if it's expanded by the SDK.
[i]It depends on how the actual expansion is performed. If it is via a 'native' coprocess (e.g. invoking java to run a server based on a bundled jar), I don't think such a thing is likely. If via a container, it is possible.
[j]The downside of that capability is that bugs (or semantic changes) may be introduced as well, unless we deal with transform versioning and let users pick a specific version.
[k]Additionally, it is not clear to the end user whether a transform will be cross language or not (because it is hidden behind a shim). For example a new end user would have to either install all target dependencies first. Or, they will discover target dependencies as they use new cross language transform and install one by one as they hit them.
This also makes it unclear what is the complete stand alone product for non-Java SDKs? Is it the SDK itself or SDK + Java dependencies?
[l]The SDK should be capable of constructing a pipeline that can be accepted by the Runner API without using a cross-language transforms (via https://s.apache.org/beam-job-api). However, connectors and other utilities may require additional dependencies.
[m]Added another SDK disadvantage
[n]This is a very high burden for Python SDK users. Not all users of this SDK are familiar with other languages/dependencies. The complexity here increases from having a virtual environment to installing functional target SDKs + languages. The former requirement (virtual environment) is already high for some users.
If we choose this path and increase the complexity, we would end up offering python version of most Java sources/transforms in order to make it easier for the Python SDK users. This will already beat the first two use cases listed above.
[o]Perhaps the party serving the expansion API for a language can be made responsible for dependencies.
[p]Kenn, could you explain how that will work?
[q]Is not this a reasonable expectation for a cross language transform author? The least, I would expect the transform authors to test/certify that their transforms work in other languages for any of the solutions in this document.
[r]I don't think it's reasonable. PTransforms are not cross-language, pipelines are. With the Fn API one should only have to assert that one's code works in the specified environment.
If an SDK in language X wants to provide convenient wrappers for transforms in language Y, yes, these should be tested. But even in this case I think there should be a single source of truth (which may be precompiled/cached to avoid construction-time dependencies).
(Also, in practice the tf.transform authors are not going to be testing everything they do on Java and Go and ..., nor is the diverse set of folks writing arcane Java connectors going to test on Python, ...)
[s]This seems as though it would be manageable for very simple transforms in which the majority of the complexity is within the UDF - e.g. many IO's can be a single transform that produces (to-be-parsed) bytes, and that can be the extent of the transform. This should largely be maintainable; however, anything more complex is unlikely to scale.
[t]Could you expand this?
[u]Done. I believe that currently we require an environment, and the choice would be embedding that environment or adding new APIs.
[v]I agree that this is a disadvantage.
I would guess that the complexity would not be high. Assuming that SDKs could all generate runner api compatible graphs, this would be a matter creating partial graphs and packaging at release time.
[w]+1, complexity is not that high here (though there's some API questions as to how to pass/interpret the payload, the output should just be a graph fragment).
[x]Though this boils down to the equivalent question of "registering" a PTransform with the above RPC API.
[y]Why?
[z]A cross-language transform author now needs to also bundle all of their associated compiled forms, so it makes distribution more complex.
[aa]I think it's a tradeoff between this or an RPC, one that each SDK can make given its target audience.
[ab]expand?
[ac]_Marked as resolved_
[ad]_Re-opened_
I wasn't following what you meant by "blessed."
[ae]those provided by a user library (something outside of the SDK)