On Tue, May 7, 2019 at 10:21 AM Maximilian Michels <m...@apache.org> wrote:
> Here's the first draft: > > https://docs.google.com/document/d/1XaiNekAY2sptuQRIXpjGAyaYdSc-wlJ-VKjl04c8N48/edit?usp=sharing > > It's rather high-level. We may want to add more details once we have > finalized the design. Feel free to make comments and edits. > Thanks Max. Added some comments. > > > All of this goes back to the idea that I think the listing of > > artifacts (or more general dependencies) should be a property of the > > environment themselves. > > +1 I came to the same conclusion while thinking about how to store > artifact information for deferred execution of the pipeline. > > -Max > > On 07.05.19 18:10, Robert Bradshaw wrote: > > Looking forward to your writeup, Max. In the meantime, some comments > below. > > > > > > From: Lukasz Cwik <lc...@google.com> > > Date: Thu, May 2, 2019 at 6:45 PM > > To: dev > > > >> > >> > >> On Thu, May 2, 2019 at 7:20 AM Robert Bradshaw <rober...@google.com> > wrote: > >>> > >>> On Sat, Apr 27, 2019 at 1:14 AM Lukasz Cwik <lc...@google.com> wrote: > >>>> > >>>> We should stick with URN + payload + artifact metadata[1] where the > only mandatory one that all SDKs and expansion services understand is the > "bytes" artifact type. This allows us to add optional URNs for file://, > http://, Maven, PyPi, ... in the future. I would make the artifact > staging service use the same URN + payload mechanism to get compatibility > of artifacts across the different services and also have the artifact > staging service be able to be queried for the list of artifact types it > supports. > >>> > >>> +1 > >>> > >>>> Finally, we would need to have environments enumerate the artifact > types that they support. > >>> > >>> Meaning at runtime, or as another field statically set in the proto? > >> > >> > >> I don't believe runners/SDKs should have to know what artifacts each > environment supports at runtime and instead have environments enumerate > them explicitly in the proto. I have been thinking about a more general > "capabilities" block on environments which allow them to enumerate URNs > that the environment understands. This would include artifact type URNs, > PTransform URNs, coder URNs, ... I haven't proposed anything specific down > this line yet because I was wondering how environment resources (CPU, min > memory, hardware like GPU, AWS/GCP/Azure/... machine types) should/could > tie into this. > >> > >>> > >>>> Having everyone have the same "artifact" representation would be > beneficial since: > >>>> a) Python environments could install dependencies from a > requirements.txt file (something that the Google Cloud Dataflow Python > docker container allows for today) > >>>> b) It provides an extensible and versioned mechanism for SDKs, > environments, and artifact staging/retrieval services to support additional > artifact types > >>>> c) Allow for expressing a canonical representation of an artifact > like a Maven package so a runner could merge environments that the runner > deems compatible. > >>>> > >>>> The flow I could see is: > >>>> 1) (optional) query artifact staging service for supported artifact > types > >>>> 2) SDK request expansion service to expand transform passing in a > list of artifact types the SDK and artifact staging service support, the > expansion service returns a list of artifact types limited to those > supported types + any supported by the environment > >>> > >>> The crux of the issue seems to be how the expansion service returns > >>> the artifacts themselves. Is this going with the approach that the > >>> caller of the expansion service must host an artifact staging service? > >> > >> > >> The caller would not need to host an artifact staging service (but > would become effectively a proxy service, see my comment below for more > details) as I would have expected this to be part of the expansion service > response. > >> > >>> > >>> There is also the question here is how the returned artifacts get > >>> attached to the various environments, or whether they get implicitly > >>> applied to all returned stages (which need not have a consistent > >>> environment)? > >> > >> > >> I would suggest returning additional information that says what > artifact is for which environment. Applying all artifacts to all > environments is likely to cause issues since some environments may not > understand certain artifact types or may get conflicting versions of > artifacts. I would see this happening since an expansion service that > aggregates other expansion services seems likely, for example: > >> /-> ExpansionSerivce(Python) > >> ExpansionService(Aggregator) --> ExpansionService(Java) > >> \-> ExpansionSerivce(Go) > > > > All of this goes back to the idea that I think the listing of > > artifacts (or more general dependencies) should be a property of the > > environment themselves. > > > >>>> 3) SDK converts any artifact types that the artifact staging service > or environment doesn't understand, e.g. pulls down Maven dependencies and > converts them to "bytes" artifacts > >>> > >>> Here I think we're conflating two things. The "type" of an artifact is > >>> both (1) how to fetch the bytes and (2) how to interpret them (e.g. is > >>> this a jar file, or a pip tarball, or just some data needed by a DoFn, > >>> or ...) Only (1) can be freely transmuted. > >> > >> > >> Your right. Thinking about this some more, general artifact conversion > is unlikely to be practical because how to interpret an artifact is > environment dependent. For example, a requirements.txt used to install pip > packages for a Python docker container depends on the filesystem layout of > that specific docker container. One could simulate doing a pip install on > the same filesystem, see the diff and then of all the packages in > requirements.txt but this quickly becomes impractical. > >> > >>> > >>>> 4) SDK sends artifacts to artifact staging service > >>>> 5) Artifact staging service converts any artifacts to types that the > environment understands > >>>> 6) Environment is started and gets artifacts from the artifact > retrieval service. > >>>> > >>>> On Wed, Apr 24, 2019 at 4:44 AM Robert Bradshaw <rober...@google.com> > wrote: > >>>>> > >>>>> On Wed, Apr 24, 2019 at 12:21 PM Maximilian Michels <m...@apache.org> > wrote: > >>>>>> > >>>>>> Good idea to let the client expose an artifact staging service that > the > >>>>>> ExpansionService could use to stage artifacts. This solves two > problems: > >>>>>> > >>>>>> (1) The Expansion Service not being able to access the Job Server > >>>>>> artifact staging service > >>>>>> (2) The client not having access to the dependencies returned by the > >>>>>> Expansion Server > >>>>>> > >>>>>> The downside is that it adds an additional indirection. The > alternative > >>>>>> to let the client handle staging the artifacts returned by the > Expansion > >>>>>> Server is more transparent and easier to implement. > >>>>> > >>>>> The other downside is that it may not always be possible for the > >>>>> expansion service to connect to the artifact staging service (e.g. > >>>>> when constructing a pipeline locally against a remote expansion > >>>>> service). > >>>> > >>>> Just to make sure, your saying the expansion service would return all > the artifacts (bytes, urls, ...) as part of the response since the > expansion service wouldn't be able to connect to the SDK that is running > locally either. > >>> > >>> Yes. Well, more I'm asking how the expansion service would return any > >>> artifacts. > >>> > >>> What we have is > >>> > >>> Runner <--- SDK ---> Expansion service. > >>> > >>> Where the unidirectional arrow means "instantiates a connection with" > >>> and the other direction (and missing arrows) may not be possible. > >> > >> > >> I believe the ExpansionService Expand request should become a > unidirectional stream back to the caller so that artifacts could be sent > back to the SDK (effectively mirroring the artifact staging service API). > So the expansion response would stream back a bunch artifact data messages > and also the expansion response containing PTransform information. > > > > +1. > > > >>>>>> Ideally, the Expansion Service won't return any dependencies > because the > >>>>>> environment already contains the required dependencies. We could > make it > >>>>>> a requirement for the expansion to be performed inside an > environment. > >>>>>> Then we would already ensure during expansion time that the runtime > >>>>>> dependencies are available. > >>>>> > >>>>> Yes, it's cleanest if the expansion service provides an environment > >>>>> without all the dependencies provided. Interesting idea to make this > a > >>>>> property of the expansion service itself. > >>>> > >>>> I had thought this too but an opaque docker container that was built > on top of a base Beam docker container would be very difficult for a runner > to introspect and check to see if its compatible to allow for fusion across > PTransforms. I think artifacts need to be communicated in their canonical > representation. > >>> > >>> It's clean (from the specification point of view), but doesn't allow > >>> for good introspection/fusion (aside from one being a base of another, > >>> perhaps). > >>> > >>>>>>> In this case, the runner would (as > >>>>>>> requested by its configuration) be free to merge environments it > >>>>>>> deemed compatible, including swapping out beam-java-X for > >>>>>>> beam-java-embedded if it considers itself compatible with the > >>>>>>> dependency list. > >>>>>> > >>>>>> Could you explain how that would work in practice? > >>>>> > >>>>> Say one has a pipeline with environments > >>>>> > >>>>> A: beam-java-sdk-2.12-docker > >>>>> B: beam-java-sdk-2.12-docker + dep1 > >>>>> C: beam-java-sdk-2.12-docker + dep2 > >>>>> D: beam-java-sdk-2.12-docker + dep3 > >>>>> > >>>>> A runner could (conceivably) be intelligent enough to know that dep1 > >>>>> and dep2 are indeed compatible, and run A, B, and C in a single > >>>>> beam-java-sdk-2.12-docker + dep1 + dep2 environment (with the > >>>>> corresponding fusion and lower overhead benefits). If a certain > >>>>> pipeline option is set, it might further note that dep1 and dep2 are > >>>>> compatible with its own workers, which are build against sdk-2.12, > and > >>>>> choose to run these in embedded + dep1 + dep2 environment. > >>>> > >>>> We have been talking about the expansion service and cross language > transforms a lot lately but I believe it will initially come at the cost of > poor fusion of transforms since "merging" environments that are compatible > is a difficult problem since it brings up many of the dependency management > issues (e.g. diamond dependency issues). > >>> > >>> I agree. I think expansion services offering "kitchen-sink" > >>> containers, when possible, can go far here. If we could at least > >>> recognize when one environment/set of deps is a superset of another, > >>> that could be an easy case that would yield a lot of benefit as well. > >> > >> > >> +1 >
