On Fri, Feb 1, 2019 at 6:12 AM Maximilian Michels <m...@apache.org> wrote:
> Yes, I imagine sources to implement a JsonConfigurable interface (e.g.
> on their builders):
>
> JsonConfigurable {
> // Either a json string or Map<String, String>
> apply(String jsonConfig);
> }
>
> In Python we would create this transform:
>
> URN: JsonConfiguredSource:v1
> payload: {
> environment: environment_id, // Java/Python/Go
> resourceIdentifier: string, // "org.apache.beam.io.PubSubIO"
> configuration: json config, // { "topic" : "my_pubsub_topic" }
> }
>
> That's more generic and could be used for other languages where we might
> have sources/sinks.
>
Looks good!
I imagine there would be a wrapper for this similar to:
https://github.com/apache/beam/blob/master/sdks/python/apache_beam/io/flink/flink_streaming_impulse_source.py#L33
> > (FWIW, I was imagining PubSubIO already had a translation into BeamFnApi
> protos that fully specified it, and we use that same format to translate
> back out.)
>
> Not that I know of.
>
> On 01.02.19 14:02, Robert Bradshaw wrote:
> > Are you suggesting something akin to a generic
> >
> > urn: JsonConfiguredJavaSource
> > payload: some json specifying which source and which parameters
> >
> > which would expand to actually constructing and applying that source?
> >
> > (FWIW, I was imagining PubSubIO already had a translation into BeamFnApi
> > protos that fully specified it, and we use that same format to translate
> > back out.)
> >
> > On Fri, Feb 1, 2019 at 1:44 PM Maximilian Michels <m...@apache.org
> > <mailto:m...@apache.org>> wrote:
> >
> > Recaping here:
> >
> > We all agree that SDF is the way to go for future implementations of
> > sources. It enables us to get rid of the source interfaces. However,
> > SDF
> > does not solve the lack of streaming sources in Python.
> >
> > The expansion PR (thanks btw!) solves the problem of
> > expanding/translating URNs known to an ExpansionService. That is a
> more
> > programmatic way of replacing language-specific performs, instead of
> > relying on translators directly in the Runner.
> >
> > What is unsolved is the configuration of sources from a foreign
> > environment. In my opinion this is the most pressing issue for Python
> > sources, because what is PubSubIO worth in Python if you cannot
> > configure it?
> >
> > What about this:
> >
> > I think it is worth adding a JSON configuration option for all
> existing
> > Java sources. That way, we could easily configure them as part of the
> > expansion request (which would contain a JSON configuration). I'll
> > probably fork a thread to discuss this in more detail, but would
> > like to
> > hear your thoughts.
> >
> > -Max
> >
> > On 01.02.19 13:08, Robert Bradshaw wrote:
> > > On Thu, Jan 31, 2019 at 6:25 PM Maximilian Michels
> > <m...@apache.org <mailto:m...@apache.org>
> > > <mailto:m...@apache.org <mailto:m...@apache.org>>> wrote:
> > >
> > > Ah, I thought you meant native Flink transforms.
> > >
> > > Exactly! The translation code is already there. The main
> > challenge
> > > is how to
> > > programmatically configure the BeamIO from Python. I suppose
> > that is
> > > also an
> > > unsolved problem for cross-language transforms in general.
> > >
> > >
> > > This is what https://github.com/apache/beam/pull/7316 does.
> > >
> > > For a particular source, one would want to define a URN and
> > > corresponding payload, then (probably) a CompositeTransform in
> > Python
> > > that takes the users arguments, packages them into the payload,
> > applies
> > > the ExternalTransform, and returns the results. How to handle
> > arbitrary
> > > UDFs embedded in sources is still TBD.
> > >
> > > For Matthias' pipeline with PubSubIO we can build something
> > > specific, but for
> > > the general case there should be way to initialize a Beam IO
> > via a
> > > configuration
> > > map provided by an external environment.
> > >
> > >
> > > I thought quite a bit about how we could represent expansions
> > statically
> > > (e.g. have some kind of expansion template that could be used, at
> > least
> > > in many cases, as data without firing up a separate process. May
> be
> > > worth doing eventually, but we run into the same issues that were
> > > discussed at
> > > https://github.com/apache/beam/pull/7316#discussion_r249996455 ).
> > >
> > > If one is already using a portable runner like Flink, having the
> job
> > > service process automatically also serve up an expansion service
> for
> > > various URNs it knows and cares about is probably a pretty low
> bar.
> > > Flink could serve up things it would rather get back untouched in
> a
> > > transform with a special flink runner urn.
> > >
> > > As Ahmet mentions, SDF is better solution. I hope it's not that
> far
> > > away, but even once it comes we'll likely want the above
> > framework to
> > > invoke the full suite of Java IOs even after they're running on
> SDF
> > > themselves.
> > >
> > > - Robert
> > >
> > > On 31.01.19 17:36, Thomas Weise wrote:
> > > > Exactly, that's what I had in mind.
> > > >
> > > > A Flink runner native transform would make the existing
> > unbounded
> > > sources
> > > > available, similar to:
> > > >
> > > >
> > >
> >
> https://github.com/apache/beam/blob/2e89c1e4d35e7b5f95a622259d23d921c3d6ad1f/runners/flink/src/main/java/org/apache/beam/runners/flink/FlinkStreamingTransformTranslators.java#L167
> > > >
> > > >
> > > >
> > > >
> > > > On Thu, Jan 31, 2019 at 8:18 AM Maximilian Michels
> > > <m...@apache.org <mailto:m...@apache.org>
> > <mailto:m...@apache.org <mailto:m...@apache.org>>
> > > > <mailto:m...@apache.org <mailto:m...@apache.org>
> > <mailto:m...@apache.org <mailto:m...@apache.org>>>> wrote:
> > > >
> > > > Wouldn't it be even more useful for the transition
> > period if
> > > we enabled Beam IO
> > > > to be used via Flink (like in the legacy Flink
> Runner)? In
> > > this particular
> > > > example, Matthias wants to use PubSubIO, which is not
> even
> > > available as a
> > > > native
> > > > Flink transform.
> > > >
> > > > On 31.01.19 16:21, Thomas Weise wrote:
> > > > > Until SDF is supported, we could also add Flink
> runner
> > > native transforms for
> > > > > selected unbounded sources [1].
> > > > >
> > > > > That might be a reasonable option to unblock users
> that
> > > want to try Python
> > > > > streaming on Flink.
> > > > >
> > > > > Thomas
> > > > >
> > > > > [1]
> > > > >
> > > >
> > >
> >
> https://github.com/lyft/beam/blob/release-2.10.0-lyft/runners/flink/src/main/java/org/apache/beam/runners/flink/LyftFlinkStreamingPortableTranslations.java
> > > > >
> > > > >
> > > > > On Thu, Jan 31, 2019 at 6:51 AM Maximilian Michels
> > > <m...@apache.org <mailto:m...@apache.org>
> > <mailto:m...@apache.org <mailto:m...@apache.org>>
> > > > <mailto:m...@apache.org <mailto:m...@apache.org>
> > <mailto:m...@apache.org <mailto:m...@apache.org>>>
> > > > > <mailto:m...@apache.org <mailto:m...@apache.org>
> > <mailto:m...@apache.org <mailto:m...@apache.org>>
> > > <mailto:m...@apache.org <mailto:m...@apache.org>
> > <mailto:m...@apache.org <mailto:m...@apache.org>>>>> wrote:
> > > > >
> > > > > > I have a hard time to imagine how can we map
> > in a
> > > generic way
> > > > > RestrictionTrackers into the existing
> > > Bounded/UnboundedSource, so I would
> > > > > love to hear more about the details.
> > > > >
> > > > > Isn't it the other way around? The SDF is a
> > > generalization of
> > > > UnboundedSource.
> > > > > So we would wrap UnboundedSource using SDF. I'm
> not
> > > saying it is
> > > > trivial, but
> > > > > SDF offers all the functionality that
> > UnboundedSource
> > > needs.
> > > > >
> > > > > For example, the @GetInitialRestriction method
> > would
> > > call split on the
> > > > > UnboundedSource and the restriction trackers
> would
> > > then be used to
> > > > process the
> > > > > splits.
> > > > >
> > > > > On 31.01.19 15:16, Ismaël Mejía wrote:
> > > > > >> Not necessarily. This would be one way.
> Another
> > > way is build an SDF
> > > > > wrapper for UnboundedSource. Probably the
> > easier path
> > > for migration.
> > > > > >
> > > > > > That would be fantastic, I have heard about
> such
> > > wrapper multiple
> > > > > > times but so far there is not any realistic
> > > proposal. I have a hard
> > > > > > time to imagine how can we map in a generic
> way
> > > RestrictionTrackers
> > > > > > into the existing Bounded/UnboundedSource,
> so I
> > > would love to hear
> > > > > > more about the details.
> > > > > >
> > > > > > On Thu, Jan 31, 2019 at 3:07 PM Maximilian
> > Michels
> > > <m...@apache.org <mailto:m...@apache.org>
> > <mailto:m...@apache.org <mailto:m...@apache.org>>
> > > > <mailto:m...@apache.org <mailto:m...@apache.org>
> > <mailto:m...@apache.org <mailto:m...@apache.org>>>
> > > > > <mailto:m...@apache.org <mailto:m...@apache.org>
> > <mailto:m...@apache.org <mailto:m...@apache.org>>
> > > <mailto:m...@apache.org <mailto:m...@apache.org>
> > <mailto:m...@apache.org <mailto:m...@apache.org>>>>> wrote:
> > > > > >>
> > > > > >> > In addition to have support in the
> runners,
> > > this will require a
> > > > > >> > rewrite of PubsubIO to use the new SDF
> API.
> > > > > >>
> > > > > >> Not necessarily. This would be one way.
> Another
> > > way is build an SDF
> > > > > wrapper for
> > > > > >> UnboundedSource. Probably the easier path
> for
> > > migration.
> > > > > >>
> > > > > >> On 31.01.19 14:03, Ismaël Mejía wrote:
> > > > > >>>> Fortunately, there is already a pending
> > PR for
> > > cross-language
> > > > > pipelines which
> > > > > >>>> will allow us to use Java IO like PubSub
> in
> > > Python jobs.
> > > > > >>>
> > > > > >>> In addition to have support in the
> > runners, this
> > > will require a
> > > > > >>> rewrite of PubsubIO to use the new SDF API.
> > > > > >>>
> > > > > >>> On Thu, Jan 31, 2019 at 12:23 PM
> > Maximilian Michels
> > > > <m...@apache.org <mailto:m...@apache.org>
> > <mailto:m...@apache.org <mailto:m...@apache.org>>
> > > <mailto:m...@apache.org <mailto:m...@apache.org>
> > <mailto:m...@apache.org <mailto:m...@apache.org>>>
> > > > > <mailto:m...@apache.org <mailto:m...@apache.org>
> > <mailto:m...@apache.org <mailto:m...@apache.org>>
> > > <mailto:m...@apache.org <mailto:m...@apache.org>
> > <mailto:m...@apache.org <mailto:m...@apache.org>>>>> wrote:
> > > > > >>>>
> > > > > >>>> Hi Matthias,
> > > > > >>>>
> > > > > >>>> This is already reflected in the
> > compatibility
> > > matrix, if you look
> > > > > under SDF.
> > > > > >>>> There is no UnboundedSource interface for
> > > portable pipelines.
> > > > That's a
> > > > > legacy
> > > > > >>>> abstraction that will be replaced with
> SDF.
> > > > > >>>>
> > > > > >>>> Fortunately, there is already a pending
> > PR for
> > > cross-language
> > > > > pipelines which
> > > > > >>>> will allow us to use Java IO like PubSub
> in
> > > Python jobs.
> > > > > >>>>
> > > > > >>>> Thanks,
> > > > > >>>> Max
> > > > > >>>>
> > > > > >>>> On 31.01.19 12:06, Matthias Baetens wrote:
> > > > > >>>>> Hey Ankur,
> > > > > >>>>>
> > > > > >>>>> Thanks for the swift reply. Should I
> change
> > > this in the
> > > > capability matrix
> > > > > >>>>>
> > > <https://s.apache.org/apache-beam-portability-support-table>
> > then?
> > > > > >>>>>
> > > > > >>>>> Many thanks.
> > > > > >>>>> Best,
> > > > > >>>>> Matthias
> > > > > >>>>>
> > > > > >>>>> On Thu, 31 Jan 2019 at 09:31, Ankur
> Goenka
> > > <goe...@google.com <mailto:goe...@google.com>
> > <mailto:goe...@google.com <mailto:goe...@google.com>>
> > > > <mailto:goe...@google.com <mailto:goe...@google.com>
> > <mailto:goe...@google.com <mailto:goe...@google.com>>>
> > > > > <mailto:goe...@google.com
> > <mailto:goe...@google.com> <mailto:goe...@google.com
> > <mailto:goe...@google.com>>
> > > <mailto:goe...@google.com <mailto:goe...@google.com>
> > <mailto:goe...@google.com <mailto:goe...@google.com>>>>
> > > > > >>>>> <mailto:goe...@google.com
> > <mailto:goe...@google.com>
> > > <mailto:goe...@google.com <mailto:goe...@google.com>>
> > <mailto:goe...@google.com <mailto:goe...@google.com>
> > > <mailto:goe...@google.com <mailto:goe...@google.com>>>
> > > > <mailto:goe...@google.com <mailto:goe...@google.com>
> > <mailto:goe...@google.com <mailto:goe...@google.com>>
> > > <mailto:goe...@google.com <mailto:goe...@google.com>
> > <mailto:goe...@google.com <mailto:goe...@google.com>>>>>> wrote:
> > > > > >>>>>
> > > > > >>>>> Hi Matthias,
> > > > > >>>>>
> > > > > >>>>> Unfortunately, unbounded reads
> > including
> > > pubsub are not yet
> > > > > supported for
> > > > > >>>>> portable runners.
> > > > > >>>>>
> > > > > >>>>> Thanks,
> > > > > >>>>> Ankur
> > > > > >>>>>
> > > > > >>>>> On Thu, Jan 31, 2019 at 2:44 PM
> > Matthias
> > > Baetens
> > > > > <baetensmatth...@gmail.com
> > <mailto:baetensmatth...@gmail.com>
> > > <mailto:baetensmatth...@gmail.com
> > <mailto:baetensmatth...@gmail.com>>
> > <mailto:baetensmatth...@gmail.com <mailto:baetensmatth...@gmail.com>
> > > <mailto:baetensmatth...@gmail.com
> > <mailto:baetensmatth...@gmail.com>>>
> > > > <mailto:baetensmatth...@gmail.com
> > <mailto:baetensmatth...@gmail.com>
> > > <mailto:baetensmatth...@gmail.com
> > <mailto:baetensmatth...@gmail.com>>
> > <mailto:baetensmatth...@gmail.com <mailto:baetensmatth...@gmail.com>
> > > <mailto:baetensmatth...@gmail.com
> > <mailto:baetensmatth...@gmail.com>>>>
> > > > > >>>>> <mailto:baetensmatth...@gmail.com
> > <mailto:baetensmatth...@gmail.com>
> > > <mailto:baetensmatth...@gmail.com
> > <mailto:baetensmatth...@gmail.com>>
> > > > <mailto:baetensmatth...@gmail.com
> > <mailto:baetensmatth...@gmail.com>
> > > <mailto:baetensmatth...@gmail.com
> > <mailto:baetensmatth...@gmail.com>>>
> > > > > <mailto:baetensmatth...@gmail.com
> > <mailto:baetensmatth...@gmail.com>
> > > <mailto:baetensmatth...@gmail.com
> > <mailto:baetensmatth...@gmail.com>>
> > > > <mailto:baetensmatth...@gmail.com
> > <mailto:baetensmatth...@gmail.com>
> > > <mailto:baetensmatth...@gmail.com
> > <mailto:baetensmatth...@gmail.com>>>>>> wrote:
> > > > > >>>>>
> > > > > >>>>> Hi everyone,
> > > > > >>>>>
> > > > > >>>>> Last few days I have been
> > trying to
> > > run a streaming
> > > > > pipeline (code on
> > > > > >>>>> Github
> > > <https://github.com/matthiasa4/beam-demo>) on a
> > > > > Flink Runner.
> > > > > >>>>>
> > > > > >>>>> I am running a Flink cluster
> > locally
> > > (v1.5.6
> > > > > >>>>>
> > > <https://flink.apache.org/downloads.html>)
> > > > > >>>>> I have built the SDK Harness
> > > Container: /./gradlew
> > > > > >>>>>
> > :beam-sdks-python-container:docker/
> > > > > >>>>> and started the JobServer:
> > /./gradlew
> > > > > >>>>>
> > > :beam-runners-flink_2.11-job-server:runShadow
> > > > > >>>>>
> -PflinkMasterUrl=localhost:8081./
> > > > > >>>>>
> > > > > >>>>> I run my pipeline with:
> > /env/bin/python
> > > > streaming_pipeline.py
> > > > > >>>>> --runner=PortableRunner
> > > --job_endpoint=localhost:8099
> > > > > --output xxx
> > > > > >>>>> --input_subscription xxx
> > > --output_subscription xxx/
> > > > > >>>>> /
> > > > > >>>>> /
> > > > > >>>>> All this is running inside a
> > Ubuntu
> > > (Bionic) in a
> > > > Virtualbox.
> > > > > >>>>>
> > > > > >>>>> The job submits fine, but
> > > unfortunately fails after
> > > > a few
> > > > > seconds with
> > > > > >>>>> the error attached.
> > > > > >>>>>
> > > > > >>>>> Anything I am missing or doing
> > wrong?
> > > > > >>>>>
> > > > > >>>>> Many thanks.
> > > > > >>>>> Best,
> > > > > >>>>> Matthias
> > > > > >>>>>
> > > > > >>>>>
> > > > >
> > > >
> > >
> >
>
