I believe the flag was never relevant for PortableRunner. I might be
wrong as well. The flag affects a few bits in the core code and that is
why the solution cannot be by just setting the flag in Dataflow runner.
It requires some amount of clean up. I agree that it would be good to
clean this up, and I also agree to not rush this especially if this is
not currently impacting users.
Ahmet
On Wed, Sep 18, 2019 at 12:56 PM Maximilian Michels <m...@apache.org
<mailto:m...@apache.org>> wrote:
> I disagree that this flag is obsolete. It is still serving a
purpose for batch users using dataflow runner and that is decent
chunk of beam python users.
It is obsolete for the PortableRunner. If the Dataflow Runner needs
this
flag, couldn't we simply add it there? As far as I know Dataflow users
do not use the PortableRunner. I might be wrong.
As Kyle mentioned, he already fixed the issue. The fix is only present
in the 2.16.0 release though. This flag has repeatedly caused friction
for users and that's why I want to get rid of it.
There is of course no need to rush this but it would be great to tackle
this for the next release. Filed a JIRA:
https://jira.apache.org/jira/browse/BEAM-8274
Cheers,
Max
On 17.09.19 15:39, Kyle Weaver wrote:
> Actually, the reported issues are already fixed on head. We're just
> trying to prevent similar issues in the future.
>
> Kyle Weaver | Software Engineer | github.com/ibzib
<http://github.com/ibzib>
> <http://github.com/ibzib> | kcwea...@google.com
<mailto:kcwea...@google.com> <mailto:kcwea...@google.com
<mailto:kcwea...@google.com>>
>
>
> On Tue, Sep 17, 2019 at 3:38 PM Ahmet Altay <al...@google.com
<mailto:al...@google.com>
> <mailto:al...@google.com <mailto:al...@google.com>>> wrote:
>
>
>
> On Tue, Sep 17, 2019 at 2:26 PM Maximilian Michels
<m...@apache.org <mailto:m...@apache.org>
> <mailto:m...@apache.org <mailto:m...@apache.org>>> wrote:
>
> > Is not this flag set automatically for the portable runner
>
> Yes, the flag is set automatically, but it has been broken
> before and
> likely will be again. It just adds additional complexity to
> portable
> Runners. There is no other portability API then the Fn
API. This
> flag
> historically had its justification, but seems obsolete now.
>
>
> I disagree that this flag is obsolete. It is still serving a
purpose
> for batch users using dataflow runner and that is decent chunk of
> beam python users.
>
> I agree with switching the default. I would like to give
enough time
> to decouple the flag from the core code. (With a quick search
I saw
> two instances related to Read and Create.) Have time to test
changes
> and then switch the default.
>
>
> An isinstance check might be smarter, but does not get rid of
> the root
> of the problem.
>
>
> I might be wrong, IIUC, it will temporarily resolve the reported
> issues. Is this not accurate?
>
>
> -Max
>
> On 17.09.19 14:20, Ahmet Altay wrote:
> > Could you make that change and see if it would have
addressed
> the issue
> > here?
> >
> > On Tue, Sep 17, 2019 at 2:18 PM Kyle Weaver
> <kcwea...@google.com <mailto:kcwea...@google.com>
<mailto:kcwea...@google.com <mailto:kcwea...@google.com>>
> > <mailto:kcwea...@google.com
<mailto:kcwea...@google.com> <mailto:kcwea...@google.com
<mailto:kcwea...@google.com>>>> wrote:
> >
> > The flag is automatically set, but not in a smart
way. Taking
> > another look at the code, a more resilient fix
would be
> to just
> > check if the runner isinstance of PortableRunner.
> >
> > Kyle Weaver | Software Engineer | github.com/ibzib
<http://github.com/ibzib>
> <http://github.com/ibzib>
> > <http://github.com/ibzib> | kcwea...@google.com
<mailto:kcwea...@google.com>
> <mailto:kcwea...@google.com <mailto:kcwea...@google.com>>
> > <mailto:kcwea...@google.com
<mailto:kcwea...@google.com> <mailto:kcwea...@google.com
<mailto:kcwea...@google.com>>>
> >
> >
> > On Tue, Sep 17, 2019 at 2:14 PM Ahmet Altay
> <al...@google.com <mailto:al...@google.com>
<mailto:al...@google.com <mailto:al...@google.com>>
> > <mailto:al...@google.com <mailto:al...@google.com>
<mailto:al...@google.com <mailto:al...@google.com>>>> wrote:
> >
> > Is not this flag set automatically for the
portable
> runner here
> > [1] ?
> >
> > [1]
> >
>
https://github.com/apache/beam/blob/f0aa877b8703eed4143957b4cd212aa026238a6e/sdks/python/apache_beam/pipeline.py#L160
> >
> > On Tue, Sep 17, 2019 at 2:07 PM Robert Bradshaw
> > <rober...@google.com
<mailto:rober...@google.com> <mailto:rober...@google.com
<mailto:rober...@google.com>>
> <mailto:rober...@google.com <mailto:rober...@google.com>
<mailto:rober...@google.com <mailto:rober...@google.com>>>> wrote:
> >
> > On Tue, Sep 17, 2019 at 1:43 PM Thomas Weise
> <t...@apache.org <mailto:t...@apache.org>
<mailto:t...@apache.org <mailto:t...@apache.org>>
> > <mailto:t...@apache.org
<mailto:t...@apache.org> <mailto:t...@apache.org
<mailto:t...@apache.org>>>>
> wrote:
> > >
> > > +1 for making --experiments=beam_fn_api
default.
> > >
> > > Can the Dataflow runner driver just
remove the
> setting if
> > it is not compatible?
> >
> > The tricky bit would be undoing the
differences
> in graph
> > construction
> > due to this flag flip. But I would be in
favor of
> changing
> > the default
> > (probably just removing the flag) and
moving the
> > non-portability parts
> > into the dataflow runner itself. (It looks
like
> the key
> > differences
> > here are for the Create and Read transforms.)
> >
> > > On Tue, Sep 17, 2019 at 11:33 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:
> > >>
> > >> +dev
> > >>
> > >> The beam_fn_api flag and the way it is
> automatically set
> > is error-prone.
> > >> Is there anything that prevents us from
> removing it? I
> > understand that
> > >> some Runners, e.g. Dataflow Runner
have two
> modes of
> > executing Python
> > >> pipelines (legacy and portable), but
at this
> point it
> > seems clear that
> > >> the portability mode should be the
default.
> > >>
> > >> Cheers,
> > >> Max
> > >>
> > >> On September 14, 2019 7:50:52 PM PDT,
Yu Watanabe
> > >> <yu.w.ten...@gmail.com
<mailto:yu.w.ten...@gmail.com>
> <mailto:yu.w.ten...@gmail.com
<mailto:yu.w.ten...@gmail.com>> <mailto:yu.w.ten...@gmail.com
<mailto:yu.w.ten...@gmail.com>
> <mailto:yu.w.ten...@gmail.com
<mailto:yu.w.ten...@gmail.com>>>>
> > wrote:
> > >>
> > >> Kyle
> > >>
> > >> Thank you for the assistance.
> > >>
> > >> By specifying "experiments" in
> PipelineOptions ,
> > >>
==========================================
> > >> options = PipelineOptions([
> > >>
> "--runner=FlinkRunner",
> > >>
"--flink_version=1.8",
> > >>
> > "--flink_master_url=localhost:8081",
> > >>
> "--experiments=beam_fn_api"
> > >> ])
> > >>
==========================================
> > >>
> > >> I was able to submit the job
successfully.
> > >>
> > >> [grpc-default-executor-0] INFO
> > >>
> org.apache.beam.runners.flink.FlinkJobInvoker -
> > Invoking job
> > >>
> >
>
BeamApp-ywatanabe-0915024400-8e0dc08_bc24de73-c729-41ad-ae27-35281b45feb9
> > >> [grpc-default-executor-0] INFO
> > >>
> >
>
org.apache.beam.runners.fnexecution.jobsubmission.JobInvocation -
> > >> Starting job invocation
> > >>
> >
>
BeamApp-ywatanabe-0915024400-8e0dc08_bc24de73-c729-41ad-ae27-35281b45feb9
> > >> [flink-runner-job-invoker] INFO
> > >>
> org.apache.beam.runners.flink.FlinkPipelineRunner -
> > Translating
> > >> pipeline to Flink program.
> > >> [flink-runner-job-invoker] INFO
> > >>
> >
> org.apache.beam.runners.flink.FlinkExecutionEnvironments -
> > Creating
> > >> a Batch Execution Environment.
> > >> [flink-runner-job-invoker] INFO
> > >>
> >
> org.apache.beam.runners.flink.FlinkExecutionEnvironments -
> > Using
> > >> Flink Master URL localhost:8081.
> > >> [flink-runner-job-invoker] WARN
> > >>
> >
>
org.apache.beam.runners.flink.FlinkExecutionEnvironments - No
> > >> default parallelism could be found.
> Defaulting to
> > parallelism 1.
> > >> Please set an explicit parallelism
with
> --parallelism
> > >> [flink-runner-job-invoker] INFO
> > >>
> org.apache.flink.api.java.ExecutionEnvironment - The
> > job has 0
> > >> registered types and 0 default Kryo
> serializers
> > >> [flink-runner-job-invoker] INFO
> > >>
> org.apache.flink.configuration.Configuration -
> > Config uses fallback
> > >> configuration key
> 'jobmanager.rpc.address' instead
> > of key 'rest.address'
> > >> [flink-runner-job-invoker] INFO
> > >>
org.apache.flink.runtime.rest.RestClient
> - Rest
> > client endpoint started.
> > >> [flink-runner-job-invoker] INFO
> > >>
> >
> org.apache.flink.client.program.rest.RestClusterClient -
> > Submitting
> > >> job 4e055a8878dda3f564a7b7c84d48510d
> (detached: false).
> > >>
> > >> Thanks,
> > >> Yu Watanabe
> > >>
> > >> On Sun, Sep 15, 2019 at 3:01 AM
Kyle Weaver
> > <kcwea...@google.com
<mailto:kcwea...@google.com> <mailto:kcwea...@google.com
<mailto:kcwea...@google.com>>
> <mailto:kcwea...@google.com <mailto:kcwea...@google.com>
<mailto:kcwea...@google.com <mailto:kcwea...@google.com>>>
> > >> <mailto:kcwea...@google.com
<mailto:kcwea...@google.com>
> <mailto:kcwea...@google.com <mailto:kcwea...@google.com>>
> > <mailto:kcwea...@google.com
<mailto:kcwea...@google.com>
> <mailto:kcwea...@google.com
<mailto:kcwea...@google.com>>>>> wrote:
> > >>
> > >> Try adding
> "--experiments=beam_fn_api" to your
> > pipeline options.
> > >> (This is a known issue with
Beam 2.15
> that will
> > be fixed in 2.16.)
> > >>
> > >> Kyle Weaver | Software Engineer |
> > github.com/ibzib <http://github.com/ibzib>
<http://github.com/ibzib>
> <http://github.com/ibzib>
> > >> <http://github.com/ibzib> |
> kcwea...@google.com <mailto:kcwea...@google.com>
<mailto:kcwea...@google.com <mailto:kcwea...@google.com>>
> > <mailto:kcwea...@google.com
<mailto:kcwea...@google.com>
> <mailto:kcwea...@google.com <mailto:kcwea...@google.com>>>
> > >> <mailto:kcwea...@google.com
<mailto:kcwea...@google.com>
> <mailto:kcwea...@google.com <mailto:kcwea...@google.com>>
> > <mailto:kcwea...@google.com
<mailto:kcwea...@google.com>
> <mailto:kcwea...@google.com <mailto:kcwea...@google.com>>>>
> > >>
> > >>
> > >> On Sat, Sep 14, 2019 at 12:52
AM Yu
> Watanabe
> > >> <yu.w.ten...@gmail.com
<mailto:yu.w.ten...@gmail.com>
> <mailto:yu.w.ten...@gmail.com <mailto:yu.w.ten...@gmail.com>>
> > <mailto:yu.w.ten...@gmail.com
<mailto:yu.w.ten...@gmail.com>
> <mailto:yu.w.ten...@gmail.com
<mailto:yu.w.ten...@gmail.com>>> <mailto:yu.w.ten...@gmail.com
<mailto:yu.w.ten...@gmail.com>
> <mailto:yu.w.ten...@gmail.com <mailto:yu.w.ten...@gmail.com>>
> > <mailto:yu.w.ten...@gmail.com
<mailto:yu.w.ten...@gmail.com>
> <mailto:yu.w.ten...@gmail.com
<mailto:yu.w.ten...@gmail.com>>>>> wrote:
> > >>
> > >> Hello.
> > >>
> > >> I am trying to spin up the
flink
> runner but
> > looks like data
> > >> serialization is failing.
> > >> I would like to ask for
help to
> get over
> > with this error.
> > >>
> > >>
> >
>
========================================================================
> > >> [flink-runner-job-invoker]
ERROR
> > >>
> >
>
org.apache.beam.runners.fnexecution.jobsubmission.JobInvocation
> > >> - Error during job invocation
> > >>
> >
>
BeamApp-ywatanabe-0914074210-2fcf987a_3dc1d4dc-4754-470a-9a23-eb8a68903016.
> > >>
> java.lang.IllegalArgumentException: unable
> > to deserialize
> > >> BoundedSource
> > >> at
> > >>
> >
>
org.apache.beam.sdk.util.SerializableUtils.deserializeFromByteArray(SerializableUtils.java:74)
> > >> at
> > >>
> >
>
org.apache.beam.runners.core.construction.ReadTranslation.boundedSourceFromProto(ReadTranslation.java:94)
> > >> at
> > >>
> >
>
org.apache.beam.runners.flink.FlinkBatchPortablePipelineTranslator.translateRead(FlinkBatchPortablePipelineTranslator.java:573)
> > >> at
> > >>
> >
>
org.apache.beam.runners.flink.FlinkBatchPortablePipelineTranslator.translate(FlinkBatchPortablePipelineTranslator.java:278)
> > >> at
> > >>
> >
>
org.apache.beam.runners.flink.FlinkBatchPortablePipelineTranslator.translate(FlinkBatchPortablePipelineTranslator.java:120)
> > >> at
> > >>
> >
>
org.apache.beam.runners.flink.FlinkPipelineRunner.runPipelineWithTranslator(FlinkPipelineRunner.java:84)
> > >> at
> > >>
> >
>
org.apache.beam.runners.flink.FlinkPipelineRunner.run(FlinkPipelineRunner.java:63)
> > >> at
> > >>
> >
>
org.apache.beam.runners.fnexecution.jobsubmission.JobInvocation.runPipeline(JobInvocation.java:74)
> > >> at
> > >>
> >
>
org.apache.beam.vendor.guava.v26_0_jre.com.google.common.util.concurrent.TrustedListenableFutureTask$TrustedFutureInterruptibleTask.runInterruptibly(TrustedListenableFutureTask.java:125)
> > >> at
> > >>
> >
>
org.apache.beam.vendor.guava.v26_0_jre.com.google.common.util.concurrent.InterruptibleTask.run(InterruptibleTask.java:57)
> > >> at
> > >>
> >
>
org.apache.beam.vendor.guava.v26_0_jre.com.google.common.util.concurrent.TrustedListenableFutureTask.run(TrustedListenableFutureTask.java:78)
> > >> at
> > >>
> >
>
java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)(python)
> > >> ywatanabe@debian-09-00:~$
> > >> at
> > >>
> >
>
java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
> > >> at
> > java.lang.Thread.run(Thread.java:748)
> > >> Caused by:
java.io.IOException:
> > FAILED_TO_UNCOMPRESS(5)
> > >> at
> > >>
> >
>
org.xerial.snappy.SnappyNative.throw_error(SnappyNative.java:98)
> > >> at
> > >>
> >
> org.xerial.snappy.SnappyNative.rawUncompress(Native Method)
> > >> at
> > >>
> >
> org.xerial.snappy.Snappy.rawUncompress(Snappy.java:474)
> > >> at
> >
org.xerial.snappy.Snappy.uncompress(Snappy.java:513)
> > >> at
> > >>
> >
>
org.xerial.snappy.SnappyInputStream.readFully(SnappyInputStream.java:147)
> > >> at
> > >>
> >
>
org.xerial.snappy.SnappyInputStream.readHeader(SnappyInputStream.java:99)
> > >> at
> > >>
> >
>
org.xerial.snappy.SnappyInputStream.<init>(SnappyInputStream.java:59)
> > >> at
> > >>
> >
>
org.apache.beam.sdk.util.SerializableUtils.deserializeFromByteArray(SerializableUtils.java:68)
> > >> ... 13 more
> > >>
> >
>
========================================================================
> > >>
> > >> My beam version is below.
> > >>
> > >>
> >
>
=======================================================================
> > >> (python)
> ywatanabe@debian-09-00:~$ pip3
> > freeze | grep
> > >> apache-beam
> > >> apache-beam==2.15.0
> > >>
> >
>
=======================================================================
> > >>
> > >> I have my harness
container ready
> on the
> > registry.
> > >>
> > >>
> >
>
=======================================================================
> > >> ywatanabe@debian-09-00:~$
docker
> search
> > >>
ywatanabe-docker-apache.bintray.io/python3
<http://ywatanabe-docker-apache.bintray.io/python3>
> <http://ywatanabe-docker-apache.bintray.io/python3>
> >
<http://ywatanabe-docker-apache.bintray.io/python3>
> > >>
> >
<http://ywatanabe-docker-apache.bintray.io/python3>
> > >> NAME
DESCRIPTION
> STARS
> > >> OFFICIAL AUTOMATED
> > >> beam/python3
> 0
> > >>
> >
>
=======================================================================
> > >>
> > >> Flink is ready on separate
cluster.
> > >>
> > >>
> >
>
=======================================================================
> > >> (python)
> ywatanabe@debian-09-00:~$ ss -atunp
> > | grep 8081
> > >> tcp LISTEN 0 128
> :::8081
> > :::*
> > >>
> >
>
=======================================================================
> > >>
> > >>
> > >> My debian version.
> > >>
> > >>
> >
>
=======================================================================
> > >>
> > >> (python)
> ywatanabe@debian-09-00:~$ cat
> > /etc/debian_version
> > >> 9.11
> > >>
> >
>
=======================================================================
> > >>
> > >>
> > >> My code snippet is below.
> > >>
> > >>
> >
>
=======================================================================
> > >>
> > >> options =
PipelineOptions([
> > >>
> "--runner=FlinkRunner",
> > >>
> "--flink_version=1.8",
> > >>
"--flink_master_url=localhost:8081"
> > >> ])
> > >>
> > >> with
> beam.Pipeline(options=options) as p:
> > >>
> > >> (p |
beam.Create(["Hello
> World"]))
> > >>
> >
>
=======================================================================
> > >>
> > >>
> > >> Would there be any other
settings
> should I
> > look for ?
> > >>
> > >> Thanks,
> > >> Yu Watanabe
> > >>
> > >> --
> > >> Yu Watanabe
> > >> Weekend Freelancer who
loves to
> challenge
> > building data
> > >> platform
> > >> yu.w.ten...@gmail.com
<mailto:yu.w.ten...@gmail.com>
> <mailto:yu.w.ten...@gmail.com
<mailto:yu.w.ten...@gmail.com>> <mailto:yu.w.ten...@gmail.com
<mailto:yu.w.ten...@gmail.com>
> <mailto:yu.w.ten...@gmail.com
<mailto:yu.w.ten...@gmail.com>>>
> > <mailto:yu.w.ten...@gmail.com
<mailto:yu.w.ten...@gmail.com>
> <mailto:yu.w.ten...@gmail.com
<mailto:yu.w.ten...@gmail.com>> <mailto:yu.w.ten...@gmail.com
<mailto:yu.w.ten...@gmail.com>
> <mailto:yu.w.ten...@gmail.com
<mailto:yu.w.ten...@gmail.com>>>>
> > >> LinkedIn icon
> > <https://www.linkedin.com/in/yuwatanabe1>
> > >> Twitter icon
> <https://twitter.com/yuwtennis>
> > >>
> > >>
> > >>
> > >> --
> > >> Yu Watanabe
> > >> Weekend Freelancer who loves to
challenge
> building
> > data platform
> > >> yu.w.ten...@gmail.com
<mailto:yu.w.ten...@gmail.com>
> <mailto:yu.w.ten...@gmail.com
<mailto:yu.w.ten...@gmail.com>> <mailto:yu.w.ten...@gmail.com
<mailto:yu.w.ten...@gmail.com>
> <mailto:yu.w.ten...@gmail.com
<mailto:yu.w.ten...@gmail.com>>>
> > <mailto:yu.w.ten...@gmail.com
<mailto:yu.w.ten...@gmail.com>
> <mailto:yu.w.ten...@gmail.com
<mailto:yu.w.ten...@gmail.com>> <mailto:yu.w.ten...@gmail.com
<mailto:yu.w.ten...@gmail.com>
> <mailto:yu.w.ten...@gmail.com
<mailto:yu.w.ten...@gmail.com>>>>
> > >> LinkedIn icon
> > <https://www.linkedin.com/in/yuwatanabe1>
Twitter
> icon
> > >> <https://twitter.com/yuwtennis>
> > >>
> >
>
