On Thu, Jan 31, 2019 at 6:25 PM Maximilian Michels <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>>> 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>>>> 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>>>> 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>>>> 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>>>>> 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>>>>> 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
> > >>>>>
> > >>>>>
> >
>
