You are viewing a plain text version of this content. The canonical link for it is here.
Posted to user@beam.apache.org by Aljoscha Krettek <al...@apache.org> on 2016/07/01 14:42:37 UTC
Re: A quick demo of Apache Beam with Docker
Hi Ismael,
what are the classpath issues that you are observing?
Cheers,
Aljoscha
On Tue, 28 Jun 2016 at 18:00 Emanuele Cesena <em...@shopkick.com> wrote:
> Hi Ismael,
>
> This is great!
>
> I’ll give it a try asap (last week of Q2… pretty busy :/).
>
> For your second question, a friend of mine was working on something
> similar:
>
> https://github.com/psmiraglia/docker-flink-formula/blob/master/start-master.sh
>
> Best,
>
>
> > On Jun 28, 2016, at 7:27 AM, Ismaël Mejía <ie...@gmail.com> wrote:
> >
> > Hello,
> >
> > Emanuele, thanks for posting your example. Congratulations, you just had
> an amazing idea. First to help people to start with Beam, but also to
> provide an easy way to run/test Beam pipelines on the Flink runner. This
> second idea is really useful for me because I am running/testing ideas in
> all the runners and this is a perfect way to do it.
> >
> > I started working to make the docker image you were based on (the one on
> https://github.com/apache/flink/tree/master/flink-contrib/docker-flink
> smaller. I just created FLINK-4118 and a PR that reduces the default image
> in 460 MB.
> > https://github.com/apache/flink/pull/2176 I hope the Flink guys accept
> the changes.
> >
> > For anyone interested the final flink image is also available from my
> docker account
> >
> > docker pull iemejia/flink
> >
> > I also started a project to contrib the integration of this smaller
> version of the Flink image with Beam into Apache Beam, this probably goes
> in the same line of work of the previous email from Max.I took the freedom
> to rebase Emanuele changes into a big commit, and start working from there
> https://github.com/iemejia/incubator-beam/tree/docker-flink. I hope we
> can share our work there, of course with the people interested (e.g.
> Emanuele and Maximilian).
> >
> > Max, I have two questions:
> >
> > 1. My current approach is based on Emanuele’s idea to create an uber jar
> with the Beam SDK + the Flink Runner + all the Beam IOs and their
> dependencies (I exclude all org.apache.flink because those are provided by
> Flink). I put this big jar on $FLINK_HOME/lib and I start Flink. However I
> created a small Beam example jar and submitted it into Flink but I am
> having classpath issues. Do you have any suggestions, is there a better way
> to do this ? I suppose my approach is far from the best but I don’t know
> how Flink deals with this 'extension' cases.
> >
> > 2. I only found a way to run both the Flink’s JobManager and TaskManager
> in daemon mode. Is there an easy way to run both as normal processes? I ask
> this because the current docker image uses supervisor to keep the processes
> alive, but if we can get rid of supervisor the image will be reduced in 40
> more MB, and be really minimalistic, any ideas?
> >
> > Regards,
> > Ismael
> >
> > ps. Amit and JB, if you want I can prepare a docker image for the spark
> runner, probably using the spark-job-server image as a base, I still have
> to check how viable is this but I think is feasible.
> >
> >
> >
> > On Tue, Jun 28, 2016 at 1:39 PM, Maximilian Michels <mx...@apache.org>
> wrote:
> > Thanks for sharing Emanuele! Looking forward to providing built-in
> > Docker support in Beam.
> >
> > On Fri, Jun 24, 2016 at 9:30 AM, Amit Sela <am...@gmail.com> wrote:
> > > You're right about standalone, I know many companies (small-medium)
> > > companies that prefer spawning standalone per use case/s. I'm currently
> > > biased now towards large clusters because of my current work place ;)
> which
> > > relates better to my previous comment.
> > >
> > >
> > > On Fri, Jun 24, 2016, 03:42 Emanuele Cesena <em...@shopkick.com>
> wrote:
> > >>
> > >> Thanks Amit!
> > >>
> > >> I chose Flink because of the current capability support and for the
> nicer
> > >> front end UI, but I have nothing against Spark — actually I’m using
> Spark in
> > >> my daily job, and chances are that if we’ll use Beam, it will be on
> Spark
> > >> first.
> > >>
> > >> I can also tell you that I know of 2 instances (MemSQL, that
> distribute
> > >> its own Spark, and our parent company SK Planet in Korea) that prefer
> Spark
> > >> standalone, mostly for performance and easy of setup. So I can see a
> lot of
> > >> potential even in production environments.
> > >>
> > >> Best,
> > >>
> > >>
> > >> > On Jun 23, 2016, at 3:42 PM, Amit Sela <am...@gmail.com>
> wrote:
> > >> >
> > >> > Thanks for sharing Emanuele, I will definitely look into trying
> > >> > something like that with Spark as well :)
> > >> > While production clusters (usually) use YARN/Mesos to manage
> resources,
> > >> > this could be really great for developers to use on a virtual
> environment.
> > >> > Really interesting!
> > >> >
> > >> > On Thu, Jun 23, 2016 at 7:21 PM Emanuele Cesena <
> emanuele@shopkick.com>
> > >> > wrote:
> > >> > Thank you Aljoscha!
> > >> >
> > >> > > On Jun 23, 2016, at 1:19 AM, Aljoscha Krettek <
> aljoscha@apache.org>
> > >> > > wrote:
> > >> > >
> > >> > > It's a very nice write up indeed! Thanks for sharing. :-)
> > >> > >
> > >> > > On Thu, 23 Jun 2016 at 07:35 Jean-Baptiste Onofré <
> jb@nanthrax.net>
> > >> > > wrote:
> > >> > > Hi Emanuele,
> > >> > >
> > >> > > this is a great example !
> > >> > >
> > >> > > It shows Beam with Flink. Maybe we can enhance a bit showing how
> the
> > >> > > same pipeline can result to different docker depending of the
> backend.
> > >> > >
> > >> > > I'm working on new "concrete" Beam samples showing that:
> > >> > >
> > >> > > https://github.com/jbonofre/beam-samples
> > >> > >
> > >> > > Great work anyway !
> > >> > >
> > >> > > Regards
> > >> > > JB
> > >> > >
> > >> > > On 06/22/2016 10:18 PM, Emanuele Cesena wrote:
> > >> > > > Hi,
> > >> > > >
> > >> > > > I just published a "quick start" with Beam and wanted to share:
> > >> > > >
> > >> > > >
> https://medium.com/@ecesena/a-quick-demo-of-apache-beam-with-docker-da98b99a502a
> > >> > > >
> > >> > > > Related repos:
> > >> > > > https://github.com/ecesena/docker-beam-flink
> > >> > > > https://github.com/ecesena/beam-starter
> > >> > > >
> > >> > > > Any feedback is more than welcome!
> > >> > > >
> > >> > > > Best,
> > >> > > > E.
> > >> > > >
> > >> > >
> > >> > > --
> > >> > > Jean-Baptiste Onofré
> > >> > > jbonofre@apache.org
> > >> > > http://blog.nanthrax.net
> > >> > > Talend - http://www.talend.com
> > >> >
> > >>
> > >
> >
>
>
Re: A quick demo of Apache Beam with Docker
Posted by Jean-Baptiste Onofré <jb...@nanthrax.net>.
Hi Isma�l,
IMHO, it's a classloader issue and we would need some Maven exclusion on
dependencies. I bet we have dependencies shaded in two different versions.
I gonna take a look.
Regards
JB
On 07/07/2016 11:30 PM, Isma�l Mej�a wrote:
> \u200bBoth Max and Alojscha, thanks for writing, I have been a bit busy these
> last
> days with other Beam stuff and I have not been able to continue testing
> my Beam
> on flink integration, I will write back to you as soon as I have time
> the check
> this again, thanks.
>
--
Jean-Baptiste Onofr�
jbonofre@apache.org
http://blog.nanthrax.net
Talend - http://www.talend.com
Re: A quick demo of Apache Beam with Docker
Posted by Lukasz Cwik <lc...@google.com>.
If you find out that Flink was setting the context class loader on the
thread, then PipelineOptionsFactory could be updated to use it for proxy
generation.
On Wed, Jul 13, 2016 at 2:50 PM, Ismaël Mejía <ie...@gmail.com> wrote:
> Indeed I think it is a classloader problem, I am going to try to debug it
> with the info you sent me, and I expect also to get some help from the
> Flink guys, because I have the impression it could be something related to
> their webapp that uses a different classpath (I don't know the inner
> details of flink to check this).
>
> Thanks.
>
>
> On Wed, Jul 13, 2016 at 6:26 PM, Lukasz Cwik <lc...@google.com> wrote:
>
>> Could it be that multiple class loaders may be in play?
>>
>> PipelineOptionsFactory defaults to use the class loader that it was
>> loaded by to create the proxies which could differ from the options
>> interface class.
>>
>> Can you see who is the class loader for WordCountOptions and who is the
>> class loader for PipelineOptionsFactory?
>>
>> https://docs.oracle.com/javase/7/docs/api/java/lang/Class.html#getClassLoader()
>>
>> PipelineOptionsFactory proxy creation could be updated to use the thread
>> context class loader similar to how it does the service loading here:
>>
>> https://github.com/apache/incubator-beam/blob/master/sdks/java/core/src/main/java/org/apache/beam/sdk/options/PipelineOptionsFactory.java#L518
>>
>>
>> On Wed, Jul 13, 2016 at 12:02 PM, Ismaël Mejía <ie...@gmail.com> wrote:
>>
>>> It was already public too, I made everything public and I also tried
>>> with a jar with only the WordCount example and I had a similar exception:
>>>
>>> Caused by: java.lang.IllegalArgumentException: interface
>>> org.apache.beam.examples.WordCount$WordCountOptions is not visible from
>>> class loader
>>> at java.lang.reflect.Proxy$ProxyClassFactory.apply(Proxy.java:581)
>>> at java.lang.reflect.Proxy$ProxyClassFactory.apply(Proxy.java:557)
>>> at java.lang.reflect.WeakCache$Factory.get(WeakCache.java:230)
>>> at java.lang.reflect.WeakCache.get(WeakCache.java:127)
>>> at java.lang.reflect.Proxy.getProxyClass0(Proxy.java:419)
>>> at java.lang.reflect.Proxy.getProxyClass(Proxy.java:371)
>>> at
>>> org.apache.beam.sdk.options.PipelineOptionsFactory.validateWellFormed(PipelineOptionsFactory.java:620)
>>> at
>>> org.apache.beam.sdk.options.PipelineOptionsFactory.parseObjects(PipelineOptionsFactory.java:1374)
>>> at
>>> org.apache.beam.sdk.options.PipelineOptionsFactory.access$400(PipelineOptionsFactory.java:107)
>>> at
>>> org.apache.beam.sdk.options.PipelineOptionsFactory$Builder.as(PipelineOptionsFactory.java:292)
>>> at org.apache.beam.examples.WordCount.main(WordCount.java:191)
>>>
>>>
>>>
>>>
>>> On Wed, Jul 13, 2016 at 3:32 PM, Lukasz Cwik <lc...@google.com> wrote:
>>>
>>>> I believe I had run into this before and it was because the outer class
>>>> was not public so to make sure, you
>>>> made org.apache.beam.samples.EventsByLocation public?
>>>>
>>>> Have you tried moving
>>>> the org.apache.beam.samples.EventsByLocation$Options inner class to its own
>>>> file?
>>>>
>>>>
>>>> On Wed, Jul 13, 2016 at 4:43 AM, Ismaël Mejía <ie...@gmail.com>
>>>> wrote:
>>>>
>>>>> For context:
>>>>>
>>>>> I just changed the beam-runners-flink_2.10 pom file to add the shade
>>>>> plugin to
>>>>> pack the flink runner dependencies in an uber jar, and exclude the
>>>>> flink classes
>>>>> (that I assume are present on flink. My goal is to create a fat jar
>>>>> and add it to
>>>>> $FLINK_HOME/lib so I can deploy any pipeline on flink (with Beam
>>>>> included):
>>>>>
>>>>> <plugin>
>>>>> <groupId>org.apache.maven.plugins</groupId>
>>>>> <artifactId>maven-shade-plugin</artifactId>
>>>>> <version>2.4.1</version>
>>>>> <executions>
>>>>> <execution>
>>>>> <phase>package</phase>
>>>>> <goals>
>>>>> <goal>shade</goal>
>>>>> </goals>
>>>>> <configuration>
>>>>> <artifactSet>
>>>>> <excludes>
>>>>> <exclude>org.apache.flink:force-shading</exclude>
>>>>>
>>>>> <exclude>org.apache.flink:flink-clients_2.10</exclude>
>>>>>
>>>>> <exclude>org.apache.flink:flink-optimizer_2.10</exclude>
>>>>> <exclude>org.apache.flink:flink-java</exclude>
>>>>>
>>>>> <exclude>org.apache.flink:flink-shaded-hadoop2</exclude>
>>>>>
>>>>> <exclude>org.apache.flink:flink-streaming-java_2.10</exclude>
>>>>> <exclude>org.apache.flink:flink-core</exclude>
>>>>> <exclude>org.apache.flink:flink-annotations</exclude>
>>>>>
>>>>> <exclude>org.apache.flink:flink-runtime_2.10</exclude>
>>>>> </excludes>
>>>>> </artifactSet>
>>>>> </configuration>
>>>>> </execution>
>>>>> </executions>
>>>>>
>>>>> </plugin>
>>>>>
>>>>>
>>>>> Then I build the beam-flink-runner uber jar:
>>>>>
>>>>> mvn clean package -DskipTests
>>>>>
>>>>> And I copy the produced jar into the $FLINK_HOME/lib directory,
>>>>> however when I try to submit a
>>>>> simple test jar, via the Flink Web UI I get this exception:
>>>>>
>>>>> org.apache.flink.client.program.ProgramInvocationException: The main
>>>>> method caused an error.
>>>>> at
>>>>> org.apache.flink.client.program.PackagedProgram.callMainMethod(PackagedProgram.java:520)
>>>>> at
>>>>> org.apache.flink.client.program.PackagedProgram.invokeInteractiveModeForExecution(PackagedProgram.java:403)
>>>>> at
>>>>> org.apache.flink.client.program.OptimizerPlanEnvironment.getOptimizedPlan(OptimizerPlanEnvironment.java:80)
>>>>> at
>>>>> org.apache.flink.client.program.Client.getOptimizedPlan(Client.java:215)
>>>>> at
>>>>> org.apache.flink.runtime.webmonitor.handlers.JarActionHandler.getJobGraphAndClassLoader(JarActionHandler.java:95)
>>>>> at
>>>>> org.apache.flink.runtime.webmonitor.handlers.JarPlanHandler.handleRequest(JarPlanHandler.java:42)
>>>>> at
>>>>> org.apache.flink.runtime.webmonitor.RuntimeMonitorHandler.respondAsLeader(RuntimeMonitorHandler.java:135)
>>>>> at
>>>>> org.apache.flink.runtime.webmonitor.RuntimeMonitorHandler.channelRead0(RuntimeMonitorHandler.java:112)
>>>>> at
>>>>> org.apache.flink.runtime.webmonitor.RuntimeMonitorHandler.channelRead0(RuntimeMonitorHandler.java:60)
>>>>> at
>>>>> io.netty.channel.SimpleChannelInboundHandler.channelRead(SimpleChannelInboundHandler.java:105)
>>>>> at
>>>>> io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:339)
>>>>> at
>>>>> io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:324)
>>>>> at
>>>>> io.netty.handler.codec.http.router.Handler.routed(Handler.java:62)
>>>>> at
>>>>> io.netty.handler.codec.http.router.DualAbstractHandler.channelRead0(DualAbstractHandler.java:57)
>>>>> at
>>>>> io.netty.handler.codec.http.router.DualAbstractHandler.channelRead0(DualAbstractHandler.java:20)
>>>>> at
>>>>> io.netty.channel.SimpleChannelInboundHandler.channelRead(SimpleChannelInboundHandler.java:105)
>>>>> at
>>>>> io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:339)
>>>>> at
>>>>> io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:324)
>>>>> at
>>>>> org.apache.flink.runtime.webmonitor.HttpRequestHandler.channelRead0(HttpRequestHandler.java:104)
>>>>> at
>>>>> org.apache.flink.runtime.webmonitor.HttpRequestHandler.channelRead0(HttpRequestHandler.java:65)
>>>>> at
>>>>> io.netty.channel.SimpleChannelInboundHandler.channelRead(SimpleChannelInboundHandler.java:105)
>>>>> at
>>>>> io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:339)
>>>>> at
>>>>> io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:324)
>>>>> at
>>>>> io.netty.handler.codec.ByteToMessageDecoder.channelRead(ByteToMessageDecoder.java:242)
>>>>> at
>>>>> io.netty.channel.CombinedChannelDuplexHandler.channelRead(CombinedChannelDuplexHandler.java:147)
>>>>> at
>>>>> io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:339)
>>>>> at
>>>>> io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:324)
>>>>> at
>>>>> io.netty.channel.DefaultChannelPipeline.fireChannelRead(DefaultChannelPipeline.java:847)
>>>>> at
>>>>> io.netty.channel.nio.AbstractNioByteChannel$NioByteUnsafe.read(AbstractNioByteChannel.java:131)
>>>>> at
>>>>> io.netty.channel.nio.NioEventLoop.processSelectedKey(NioEventLoop.java:511)
>>>>> at
>>>>> io.netty.channel.nio.NioEventLoop.processSelectedKeysOptimized(NioEventLoop.java:468)
>>>>> at
>>>>> io.netty.channel.nio.NioEventLoop.processSelectedKeys(NioEventLoop.java:382)
>>>>> at io.netty.channel.nio.NioEventLoop.run(NioEventLoop.java:354)
>>>>> at
>>>>> io.netty.util.concurrent.SingleThreadEventExecutor$2.run(SingleThreadEventExecutor.java:111)
>>>>> at
>>>>> io.netty.util.concurrent.DefaultThreadFactory$DefaultRunnableDecorator.run(DefaultThreadFactory.java:137)
>>>>> at java.lang.Thread.run(Thread.java:745)
>>>>> Caused by: java.lang.IllegalArgumentException: interface
>>>>> org.apache.beam.samples.EventsByLocation$Options is not visible from class
>>>>> loader
>>>>> at java.lang.reflect.Proxy$ProxyClassFactory.apply(Proxy.java:581)
>>>>> at java.lang.reflect.Proxy$ProxyClassFactory.apply(Proxy.java:557)
>>>>> at java.lang.reflect.WeakCache$Factory.get(WeakCache.java:230)
>>>>> at java.lang.reflect.WeakCache.get(WeakCache.java:127)
>>>>> at java.lang.reflect.Proxy.getProxyClass0(Proxy.java:419)
>>>>> at java.lang.reflect.Proxy.getProxyClass(Proxy.java:371)
>>>>> at
>>>>> org.apache.beam.sdk.options.PipelineOptionsFactory.validateWellFormed(PipelineOptionsFactory.java:620)
>>>>> at
>>>>> org.apache.beam.sdk.options.PipelineOptionsFactory.parseObjects(PipelineOptionsFactory.java:1374)
>>>>> at
>>>>> org.apache.beam.sdk.options.PipelineOptionsFactory.access$400(PipelineOptionsFactory.java:107)
>>>>> at
>>>>> org.apache.beam.sdk.options.PipelineOptionsFactory$Builder.as(PipelineOptionsFactory.java:292)
>>>>> at
>>>>> org.apache.beam.samples.EventsByLocation.main(EventsByLocation.java:62)
>>>>> at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
>>>>> at
>>>>> sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
>>>>> at
>>>>> sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
>>>>> at java.lang.reflect.Method.invoke(Method.java:498)
>>>>> at
>>>>> org.apache.flink.client.program.PackagedProgram.callMainMethod(PackagedProgram.java:505)
>>>>> ... 35 more
>>>>>
>>>>> Sorry for the long exception in the message but I wanted to show all
>>>>> possible details so you can help me fix this. Notice that the interface
>>>>> org.apache.beam.samples.EventsByLocation$Options is a classical Beam
>>>>> Options interface created to configure the Pipeline, and I tried making it
>>>>> public but access does not seem to be the case, but classloading.
>>>>>
>>>>> Thanks,
>>>>> Ismael
>>>>>
>>>>> ps.
>>>>> I saw this and I think maybe it could be related:
>>>>>
>>>>> https://stackoverflow.com/questions/211176/interface-is-not-visible-from-classloader-when-using-a-proxy
>>>>>
>>>>>
>>>>> On Thu, Jul 7, 2016 at 11:30 PM, Ismaël Mejía <ie...@gmail.com>
>>>>> wrote:
>>>>>
>>>>>> Both Max and Alojscha, thanks for writing, I have been a bit busy
>>>>>> these last
>>>>>> days with other Beam stuff and I have not been able to continue
>>>>>> testing my Beam
>>>>>> on flink integration, I will write back to you as soon as I have time
>>>>>> the check
>>>>>> this again, thanks.
>>>>>>
>>>>>>
>>>>>
>>>>
>>>
>>
>
Re: A quick demo of Apache Beam with Docker
Posted by Ismaël Mejía <ie...@gmail.com>.
Indeed I think it is a classloader problem, I am going to try to debug it
with the info you sent me, and I expect also to get some help from the
Flink guys, because I have the impression it could be something related to
their webapp that uses a different classpath (I don't know the inner
details of flink to check this).
Thanks.
On Wed, Jul 13, 2016 at 6:26 PM, Lukasz Cwik <lc...@google.com> wrote:
> Could it be that multiple class loaders may be in play?
>
> PipelineOptionsFactory defaults to use the class loader that it was loaded
> by to create the proxies which could differ from the options interface
> class.
>
> Can you see who is the class loader for WordCountOptions and who is the
> class loader for PipelineOptionsFactory?
>
> https://docs.oracle.com/javase/7/docs/api/java/lang/Class.html#getClassLoader()
>
> PipelineOptionsFactory proxy creation could be updated to use the thread
> context class loader similar to how it does the service loading here:
>
> https://github.com/apache/incubator-beam/blob/master/sdks/java/core/src/main/java/org/apache/beam/sdk/options/PipelineOptionsFactory.java#L518
>
>
> On Wed, Jul 13, 2016 at 12:02 PM, Ismaël Mejía <ie...@gmail.com> wrote:
>
>> It was already public too, I made everything public and I also tried with
>> a jar with only the WordCount example and I had a similar exception:
>>
>> Caused by: java.lang.IllegalArgumentException: interface
>> org.apache.beam.examples.WordCount$WordCountOptions is not visible from
>> class loader
>> at java.lang.reflect.Proxy$ProxyClassFactory.apply(Proxy.java:581)
>> at java.lang.reflect.Proxy$ProxyClassFactory.apply(Proxy.java:557)
>> at java.lang.reflect.WeakCache$Factory.get(WeakCache.java:230)
>> at java.lang.reflect.WeakCache.get(WeakCache.java:127)
>> at java.lang.reflect.Proxy.getProxyClass0(Proxy.java:419)
>> at java.lang.reflect.Proxy.getProxyClass(Proxy.java:371)
>> at
>> org.apache.beam.sdk.options.PipelineOptionsFactory.validateWellFormed(PipelineOptionsFactory.java:620)
>> at
>> org.apache.beam.sdk.options.PipelineOptionsFactory.parseObjects(PipelineOptionsFactory.java:1374)
>> at
>> org.apache.beam.sdk.options.PipelineOptionsFactory.access$400(PipelineOptionsFactory.java:107)
>> at
>> org.apache.beam.sdk.options.PipelineOptionsFactory$Builder.as(PipelineOptionsFactory.java:292)
>> at org.apache.beam.examples.WordCount.main(WordCount.java:191)
>>
>>
>>
>>
>> On Wed, Jul 13, 2016 at 3:32 PM, Lukasz Cwik <lc...@google.com> wrote:
>>
>>> I believe I had run into this before and it was because the outer class
>>> was not public so to make sure, you
>>> made org.apache.beam.samples.EventsByLocation public?
>>>
>>> Have you tried moving
>>> the org.apache.beam.samples.EventsByLocation$Options inner class to its own
>>> file?
>>>
>>>
>>> On Wed, Jul 13, 2016 at 4:43 AM, Ismaël Mejía <ie...@gmail.com> wrote:
>>>
>>>> For context:
>>>>
>>>> I just changed the beam-runners-flink_2.10 pom file to add the shade
>>>> plugin to
>>>> pack the flink runner dependencies in an uber jar, and exclude the
>>>> flink classes
>>>> (that I assume are present on flink. My goal is to create a fat jar and
>>>> add it to
>>>> $FLINK_HOME/lib so I can deploy any pipeline on flink (with Beam
>>>> included):
>>>>
>>>> <plugin>
>>>> <groupId>org.apache.maven.plugins</groupId>
>>>> <artifactId>maven-shade-plugin</artifactId>
>>>> <version>2.4.1</version>
>>>> <executions>
>>>> <execution>
>>>> <phase>package</phase>
>>>> <goals>
>>>> <goal>shade</goal>
>>>> </goals>
>>>> <configuration>
>>>> <artifactSet>
>>>> <excludes>
>>>> <exclude>org.apache.flink:force-shading</exclude>
>>>> <exclude>org.apache.flink:flink-clients_2.10</exclude>
>>>>
>>>> <exclude>org.apache.flink:flink-optimizer_2.10</exclude>
>>>> <exclude>org.apache.flink:flink-java</exclude>
>>>>
>>>> <exclude>org.apache.flink:flink-shaded-hadoop2</exclude>
>>>>
>>>> <exclude>org.apache.flink:flink-streaming-java_2.10</exclude>
>>>> <exclude>org.apache.flink:flink-core</exclude>
>>>> <exclude>org.apache.flink:flink-annotations</exclude>
>>>> <exclude>org.apache.flink:flink-runtime_2.10</exclude>
>>>> </excludes>
>>>> </artifactSet>
>>>> </configuration>
>>>> </execution>
>>>> </executions>
>>>>
>>>> </plugin>
>>>>
>>>>
>>>> Then I build the beam-flink-runner uber jar:
>>>>
>>>> mvn clean package -DskipTests
>>>>
>>>> And I copy the produced jar into the $FLINK_HOME/lib directory, however
>>>> when I try to submit a
>>>> simple test jar, via the Flink Web UI I get this exception:
>>>>
>>>> org.apache.flink.client.program.ProgramInvocationException: The main
>>>> method caused an error.
>>>> at
>>>> org.apache.flink.client.program.PackagedProgram.callMainMethod(PackagedProgram.java:520)
>>>> at
>>>> org.apache.flink.client.program.PackagedProgram.invokeInteractiveModeForExecution(PackagedProgram.java:403)
>>>> at
>>>> org.apache.flink.client.program.OptimizerPlanEnvironment.getOptimizedPlan(OptimizerPlanEnvironment.java:80)
>>>> at
>>>> org.apache.flink.client.program.Client.getOptimizedPlan(Client.java:215)
>>>> at
>>>> org.apache.flink.runtime.webmonitor.handlers.JarActionHandler.getJobGraphAndClassLoader(JarActionHandler.java:95)
>>>> at
>>>> org.apache.flink.runtime.webmonitor.handlers.JarPlanHandler.handleRequest(JarPlanHandler.java:42)
>>>> at
>>>> org.apache.flink.runtime.webmonitor.RuntimeMonitorHandler.respondAsLeader(RuntimeMonitorHandler.java:135)
>>>> at
>>>> org.apache.flink.runtime.webmonitor.RuntimeMonitorHandler.channelRead0(RuntimeMonitorHandler.java:112)
>>>> at
>>>> org.apache.flink.runtime.webmonitor.RuntimeMonitorHandler.channelRead0(RuntimeMonitorHandler.java:60)
>>>> at
>>>> io.netty.channel.SimpleChannelInboundHandler.channelRead(SimpleChannelInboundHandler.java:105)
>>>> at
>>>> io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:339)
>>>> at
>>>> io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:324)
>>>> at
>>>> io.netty.handler.codec.http.router.Handler.routed(Handler.java:62)
>>>> at
>>>> io.netty.handler.codec.http.router.DualAbstractHandler.channelRead0(DualAbstractHandler.java:57)
>>>> at
>>>> io.netty.handler.codec.http.router.DualAbstractHandler.channelRead0(DualAbstractHandler.java:20)
>>>> at
>>>> io.netty.channel.SimpleChannelInboundHandler.channelRead(SimpleChannelInboundHandler.java:105)
>>>> at
>>>> io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:339)
>>>> at
>>>> io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:324)
>>>> at
>>>> org.apache.flink.runtime.webmonitor.HttpRequestHandler.channelRead0(HttpRequestHandler.java:104)
>>>> at
>>>> org.apache.flink.runtime.webmonitor.HttpRequestHandler.channelRead0(HttpRequestHandler.java:65)
>>>> at
>>>> io.netty.channel.SimpleChannelInboundHandler.channelRead(SimpleChannelInboundHandler.java:105)
>>>> at
>>>> io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:339)
>>>> at
>>>> io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:324)
>>>> at
>>>> io.netty.handler.codec.ByteToMessageDecoder.channelRead(ByteToMessageDecoder.java:242)
>>>> at
>>>> io.netty.channel.CombinedChannelDuplexHandler.channelRead(CombinedChannelDuplexHandler.java:147)
>>>> at
>>>> io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:339)
>>>> at
>>>> io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:324)
>>>> at
>>>> io.netty.channel.DefaultChannelPipeline.fireChannelRead(DefaultChannelPipeline.java:847)
>>>> at
>>>> io.netty.channel.nio.AbstractNioByteChannel$NioByteUnsafe.read(AbstractNioByteChannel.java:131)
>>>> at
>>>> io.netty.channel.nio.NioEventLoop.processSelectedKey(NioEventLoop.java:511)
>>>> at
>>>> io.netty.channel.nio.NioEventLoop.processSelectedKeysOptimized(NioEventLoop.java:468)
>>>> at
>>>> io.netty.channel.nio.NioEventLoop.processSelectedKeys(NioEventLoop.java:382)
>>>> at io.netty.channel.nio.NioEventLoop.run(NioEventLoop.java:354)
>>>> at
>>>> io.netty.util.concurrent.SingleThreadEventExecutor$2.run(SingleThreadEventExecutor.java:111)
>>>> at
>>>> io.netty.util.concurrent.DefaultThreadFactory$DefaultRunnableDecorator.run(DefaultThreadFactory.java:137)
>>>> at java.lang.Thread.run(Thread.java:745)
>>>> Caused by: java.lang.IllegalArgumentException: interface
>>>> org.apache.beam.samples.EventsByLocation$Options is not visible from class
>>>> loader
>>>> at java.lang.reflect.Proxy$ProxyClassFactory.apply(Proxy.java:581)
>>>> at java.lang.reflect.Proxy$ProxyClassFactory.apply(Proxy.java:557)
>>>> at java.lang.reflect.WeakCache$Factory.get(WeakCache.java:230)
>>>> at java.lang.reflect.WeakCache.get(WeakCache.java:127)
>>>> at java.lang.reflect.Proxy.getProxyClass0(Proxy.java:419)
>>>> at java.lang.reflect.Proxy.getProxyClass(Proxy.java:371)
>>>> at
>>>> org.apache.beam.sdk.options.PipelineOptionsFactory.validateWellFormed(PipelineOptionsFactory.java:620)
>>>> at
>>>> org.apache.beam.sdk.options.PipelineOptionsFactory.parseObjects(PipelineOptionsFactory.java:1374)
>>>> at
>>>> org.apache.beam.sdk.options.PipelineOptionsFactory.access$400(PipelineOptionsFactory.java:107)
>>>> at
>>>> org.apache.beam.sdk.options.PipelineOptionsFactory$Builder.as(PipelineOptionsFactory.java:292)
>>>> at
>>>> org.apache.beam.samples.EventsByLocation.main(EventsByLocation.java:62)
>>>> at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
>>>> at
>>>> sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
>>>> at
>>>> sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
>>>> at java.lang.reflect.Method.invoke(Method.java:498)
>>>> at
>>>> org.apache.flink.client.program.PackagedProgram.callMainMethod(PackagedProgram.java:505)
>>>> ... 35 more
>>>>
>>>> Sorry for the long exception in the message but I wanted to show all
>>>> possible details so you can help me fix this. Notice that the interface
>>>> org.apache.beam.samples.EventsByLocation$Options is a classical Beam
>>>> Options interface created to configure the Pipeline, and I tried making it
>>>> public but access does not seem to be the case, but classloading.
>>>>
>>>> Thanks,
>>>> Ismael
>>>>
>>>> ps.
>>>> I saw this and I think maybe it could be related:
>>>>
>>>> https://stackoverflow.com/questions/211176/interface-is-not-visible-from-classloader-when-using-a-proxy
>>>>
>>>>
>>>> On Thu, Jul 7, 2016 at 11:30 PM, Ismaël Mejía <ie...@gmail.com>
>>>> wrote:
>>>>
>>>>> Both Max and Alojscha, thanks for writing, I have been a bit busy
>>>>> these last
>>>>> days with other Beam stuff and I have not been able to continue
>>>>> testing my Beam
>>>>> on flink integration, I will write back to you as soon as I have time
>>>>> the check
>>>>> this again, thanks.
>>>>>
>>>>>
>>>>
>>>
>>
>
Re: A quick demo of Apache Beam with Docker
Posted by Lukasz Cwik <lc...@google.com>.
Could it be that multiple class loaders may be in play?
PipelineOptionsFactory defaults to use the class loader that it was loaded
by to create the proxies which could differ from the options interface
class.
Can you see who is the class loader for WordCountOptions and who is the
class loader for PipelineOptionsFactory?
https://docs.oracle.com/javase/7/docs/api/java/lang/Class.html#getClassLoader()
PipelineOptionsFactory proxy creation could be updated to use the thread
context class loader similar to how it does the service loading here:
https://github.com/apache/incubator-beam/blob/master/sdks/java/core/src/main/java/org/apache/beam/sdk/options/PipelineOptionsFactory.java#L518
On Wed, Jul 13, 2016 at 12:02 PM, Ismaël Mejía <ie...@gmail.com> wrote:
> It was already public too, I made everything public and I also tried with
> a jar with only the WordCount example and I had a similar exception:
>
> Caused by: java.lang.IllegalArgumentException: interface
> org.apache.beam.examples.WordCount$WordCountOptions is not visible from
> class loader
> at java.lang.reflect.Proxy$ProxyClassFactory.apply(Proxy.java:581)
> at java.lang.reflect.Proxy$ProxyClassFactory.apply(Proxy.java:557)
> at java.lang.reflect.WeakCache$Factory.get(WeakCache.java:230)
> at java.lang.reflect.WeakCache.get(WeakCache.java:127)
> at java.lang.reflect.Proxy.getProxyClass0(Proxy.java:419)
> at java.lang.reflect.Proxy.getProxyClass(Proxy.java:371)
> at
> org.apache.beam.sdk.options.PipelineOptionsFactory.validateWellFormed(PipelineOptionsFactory.java:620)
> at
> org.apache.beam.sdk.options.PipelineOptionsFactory.parseObjects(PipelineOptionsFactory.java:1374)
> at
> org.apache.beam.sdk.options.PipelineOptionsFactory.access$400(PipelineOptionsFactory.java:107)
> at
> org.apache.beam.sdk.options.PipelineOptionsFactory$Builder.as(PipelineOptionsFactory.java:292)
> at org.apache.beam.examples.WordCount.main(WordCount.java:191)
>
>
>
>
> On Wed, Jul 13, 2016 at 3:32 PM, Lukasz Cwik <lc...@google.com> wrote:
>
>> I believe I had run into this before and it was because the outer class
>> was not public so to make sure, you
>> made org.apache.beam.samples.EventsByLocation public?
>>
>> Have you tried moving
>> the org.apache.beam.samples.EventsByLocation$Options inner class to its own
>> file?
>>
>>
>> On Wed, Jul 13, 2016 at 4:43 AM, Ismaël Mejía <ie...@gmail.com> wrote:
>>
>>> For context:
>>>
>>> I just changed the beam-runners-flink_2.10 pom file to add the shade
>>> plugin to
>>> pack the flink runner dependencies in an uber jar, and exclude the flink
>>> classes
>>> (that I assume are present on flink. My goal is to create a fat jar and
>>> add it to
>>> $FLINK_HOME/lib so I can deploy any pipeline on flink (with Beam
>>> included):
>>>
>>> <plugin>
>>> <groupId>org.apache.maven.plugins</groupId>
>>> <artifactId>maven-shade-plugin</artifactId>
>>> <version>2.4.1</version>
>>> <executions>
>>> <execution>
>>> <phase>package</phase>
>>> <goals>
>>> <goal>shade</goal>
>>> </goals>
>>> <configuration>
>>> <artifactSet>
>>> <excludes>
>>> <exclude>org.apache.flink:force-shading</exclude>
>>> <exclude>org.apache.flink:flink-clients_2.10</exclude>
>>>
>>> <exclude>org.apache.flink:flink-optimizer_2.10</exclude>
>>> <exclude>org.apache.flink:flink-java</exclude>
>>>
>>> <exclude>org.apache.flink:flink-shaded-hadoop2</exclude>
>>>
>>> <exclude>org.apache.flink:flink-streaming-java_2.10</exclude>
>>> <exclude>org.apache.flink:flink-core</exclude>
>>> <exclude>org.apache.flink:flink-annotations</exclude>
>>> <exclude>org.apache.flink:flink-runtime_2.10</exclude>
>>> </excludes>
>>> </artifactSet>
>>> </configuration>
>>> </execution>
>>> </executions>
>>>
>>> </plugin>
>>>
>>>
>>> Then I build the beam-flink-runner uber jar:
>>>
>>> mvn clean package -DskipTests
>>>
>>> And I copy the produced jar into the $FLINK_HOME/lib directory, however
>>> when I try to submit a
>>> simple test jar, via the Flink Web UI I get this exception:
>>>
>>> org.apache.flink.client.program.ProgramInvocationException: The main
>>> method caused an error.
>>> at
>>> org.apache.flink.client.program.PackagedProgram.callMainMethod(PackagedProgram.java:520)
>>> at
>>> org.apache.flink.client.program.PackagedProgram.invokeInteractiveModeForExecution(PackagedProgram.java:403)
>>> at
>>> org.apache.flink.client.program.OptimizerPlanEnvironment.getOptimizedPlan(OptimizerPlanEnvironment.java:80)
>>> at
>>> org.apache.flink.client.program.Client.getOptimizedPlan(Client.java:215)
>>> at
>>> org.apache.flink.runtime.webmonitor.handlers.JarActionHandler.getJobGraphAndClassLoader(JarActionHandler.java:95)
>>> at
>>> org.apache.flink.runtime.webmonitor.handlers.JarPlanHandler.handleRequest(JarPlanHandler.java:42)
>>> at
>>> org.apache.flink.runtime.webmonitor.RuntimeMonitorHandler.respondAsLeader(RuntimeMonitorHandler.java:135)
>>> at
>>> org.apache.flink.runtime.webmonitor.RuntimeMonitorHandler.channelRead0(RuntimeMonitorHandler.java:112)
>>> at
>>> org.apache.flink.runtime.webmonitor.RuntimeMonitorHandler.channelRead0(RuntimeMonitorHandler.java:60)
>>> at
>>> io.netty.channel.SimpleChannelInboundHandler.channelRead(SimpleChannelInboundHandler.java:105)
>>> at
>>> io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:339)
>>> at
>>> io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:324)
>>> at io.netty.handler.codec.http.router.Handler.routed(Handler.java:62)
>>> at
>>> io.netty.handler.codec.http.router.DualAbstractHandler.channelRead0(DualAbstractHandler.java:57)
>>> at
>>> io.netty.handler.codec.http.router.DualAbstractHandler.channelRead0(DualAbstractHandler.java:20)
>>> at
>>> io.netty.channel.SimpleChannelInboundHandler.channelRead(SimpleChannelInboundHandler.java:105)
>>> at
>>> io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:339)
>>> at
>>> io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:324)
>>> at
>>> org.apache.flink.runtime.webmonitor.HttpRequestHandler.channelRead0(HttpRequestHandler.java:104)
>>> at
>>> org.apache.flink.runtime.webmonitor.HttpRequestHandler.channelRead0(HttpRequestHandler.java:65)
>>> at
>>> io.netty.channel.SimpleChannelInboundHandler.channelRead(SimpleChannelInboundHandler.java:105)
>>> at
>>> io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:339)
>>> at
>>> io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:324)
>>> at
>>> io.netty.handler.codec.ByteToMessageDecoder.channelRead(ByteToMessageDecoder.java:242)
>>> at
>>> io.netty.channel.CombinedChannelDuplexHandler.channelRead(CombinedChannelDuplexHandler.java:147)
>>> at
>>> io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:339)
>>> at
>>> io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:324)
>>> at
>>> io.netty.channel.DefaultChannelPipeline.fireChannelRead(DefaultChannelPipeline.java:847)
>>> at
>>> io.netty.channel.nio.AbstractNioByteChannel$NioByteUnsafe.read(AbstractNioByteChannel.java:131)
>>> at
>>> io.netty.channel.nio.NioEventLoop.processSelectedKey(NioEventLoop.java:511)
>>> at
>>> io.netty.channel.nio.NioEventLoop.processSelectedKeysOptimized(NioEventLoop.java:468)
>>> at
>>> io.netty.channel.nio.NioEventLoop.processSelectedKeys(NioEventLoop.java:382)
>>> at io.netty.channel.nio.NioEventLoop.run(NioEventLoop.java:354)
>>> at
>>> io.netty.util.concurrent.SingleThreadEventExecutor$2.run(SingleThreadEventExecutor.java:111)
>>> at
>>> io.netty.util.concurrent.DefaultThreadFactory$DefaultRunnableDecorator.run(DefaultThreadFactory.java:137)
>>> at java.lang.Thread.run(Thread.java:745)
>>> Caused by: java.lang.IllegalArgumentException: interface
>>> org.apache.beam.samples.EventsByLocation$Options is not visible from class
>>> loader
>>> at java.lang.reflect.Proxy$ProxyClassFactory.apply(Proxy.java:581)
>>> at java.lang.reflect.Proxy$ProxyClassFactory.apply(Proxy.java:557)
>>> at java.lang.reflect.WeakCache$Factory.get(WeakCache.java:230)
>>> at java.lang.reflect.WeakCache.get(WeakCache.java:127)
>>> at java.lang.reflect.Proxy.getProxyClass0(Proxy.java:419)
>>> at java.lang.reflect.Proxy.getProxyClass(Proxy.java:371)
>>> at
>>> org.apache.beam.sdk.options.PipelineOptionsFactory.validateWellFormed(PipelineOptionsFactory.java:620)
>>> at
>>> org.apache.beam.sdk.options.PipelineOptionsFactory.parseObjects(PipelineOptionsFactory.java:1374)
>>> at
>>> org.apache.beam.sdk.options.PipelineOptionsFactory.access$400(PipelineOptionsFactory.java:107)
>>> at
>>> org.apache.beam.sdk.options.PipelineOptionsFactory$Builder.as(PipelineOptionsFactory.java:292)
>>> at
>>> org.apache.beam.samples.EventsByLocation.main(EventsByLocation.java:62)
>>> at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
>>> at
>>> sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
>>> at
>>> sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
>>> at java.lang.reflect.Method.invoke(Method.java:498)
>>> at
>>> org.apache.flink.client.program.PackagedProgram.callMainMethod(PackagedProgram.java:505)
>>> ... 35 more
>>>
>>> Sorry for the long exception in the message but I wanted to show all
>>> possible details so you can help me fix this. Notice that the interface
>>> org.apache.beam.samples.EventsByLocation$Options is a classical Beam
>>> Options interface created to configure the Pipeline, and I tried making it
>>> public but access does not seem to be the case, but classloading.
>>>
>>> Thanks,
>>> Ismael
>>>
>>> ps.
>>> I saw this and I think maybe it could be related:
>>>
>>> https://stackoverflow.com/questions/211176/interface-is-not-visible-from-classloader-when-using-a-proxy
>>>
>>>
>>> On Thu, Jul 7, 2016 at 11:30 PM, Ismaël Mejía <ie...@gmail.com> wrote:
>>>
>>>> Both Max and Alojscha, thanks for writing, I have been a bit busy
>>>> these last
>>>> days with other Beam stuff and I have not been able to continue testing
>>>> my Beam
>>>> on flink integration, I will write back to you as soon as I have time
>>>> the check
>>>> this again, thanks.
>>>>
>>>>
>>>
>>
>
Re: A quick demo of Apache Beam with Docker
Posted by Ismaël Mejía <ie...@gmail.com>.
It was already public too, I made everything public and I also tried with
a jar with only the WordCount example and I had a similar exception:
Caused by: java.lang.IllegalArgumentException: interface
org.apache.beam.examples.WordCount$WordCountOptions is not visible from
class loader
at java.lang.reflect.Proxy$ProxyClassFactory.apply(Proxy.java:581)
at java.lang.reflect.Proxy$ProxyClassFactory.apply(Proxy.java:557)
at java.lang.reflect.WeakCache$Factory.get(WeakCache.java:230)
at java.lang.reflect.WeakCache.get(WeakCache.java:127)
at java.lang.reflect.Proxy.getProxyClass0(Proxy.java:419)
at java.lang.reflect.Proxy.getProxyClass(Proxy.java:371)
at
org.apache.beam.sdk.options.PipelineOptionsFactory.validateWellFormed(PipelineOptionsFactory.java:620)
at
org.apache.beam.sdk.options.PipelineOptionsFactory.parseObjects(PipelineOptionsFactory.java:1374)
at
org.apache.beam.sdk.options.PipelineOptionsFactory.access$400(PipelineOptionsFactory.java:107)
at
org.apache.beam.sdk.options.PipelineOptionsFactory$Builder.as(PipelineOptionsFactory.java:292)
at org.apache.beam.examples.WordCount.main(WordCount.java:191)
On Wed, Jul 13, 2016 at 3:32 PM, Lukasz Cwik <lc...@google.com> wrote:
> I believe I had run into this before and it was because the outer class
> was not public so to make sure, you
> made org.apache.beam.samples.EventsByLocation public?
>
> Have you tried moving the org.apache.beam.samples.EventsByLocation$Options
> inner class to its own file?
>
>
> On Wed, Jul 13, 2016 at 4:43 AM, Ismaël Mejía <ie...@gmail.com> wrote:
>
>> For context:
>>
>> I just changed the beam-runners-flink_2.10 pom file to add the shade
>> plugin to
>> pack the flink runner dependencies in an uber jar, and exclude the flink
>> classes
>> (that I assume are present on flink. My goal is to create a fat jar and
>> add it to
>> $FLINK_HOME/lib so I can deploy any pipeline on flink (with Beam
>> included):
>>
>> <plugin>
>> <groupId>org.apache.maven.plugins</groupId>
>> <artifactId>maven-shade-plugin</artifactId>
>> <version>2.4.1</version>
>> <executions>
>> <execution>
>> <phase>package</phase>
>> <goals>
>> <goal>shade</goal>
>> </goals>
>> <configuration>
>> <artifactSet>
>> <excludes>
>> <exclude>org.apache.flink:force-shading</exclude>
>> <exclude>org.apache.flink:flink-clients_2.10</exclude>
>> <exclude>org.apache.flink:flink-optimizer_2.10</exclude>
>> <exclude>org.apache.flink:flink-java</exclude>
>> <exclude>org.apache.flink:flink-shaded-hadoop2</exclude>
>>
>> <exclude>org.apache.flink:flink-streaming-java_2.10</exclude>
>> <exclude>org.apache.flink:flink-core</exclude>
>> <exclude>org.apache.flink:flink-annotations</exclude>
>> <exclude>org.apache.flink:flink-runtime_2.10</exclude>
>> </excludes>
>> </artifactSet>
>> </configuration>
>> </execution>
>> </executions>
>>
>> </plugin>
>>
>>
>> Then I build the beam-flink-runner uber jar:
>>
>> mvn clean package -DskipTests
>>
>> And I copy the produced jar into the $FLINK_HOME/lib directory, however
>> when I try to submit a
>> simple test jar, via the Flink Web UI I get this exception:
>>
>> org.apache.flink.client.program.ProgramInvocationException: The main
>> method caused an error.
>> at
>> org.apache.flink.client.program.PackagedProgram.callMainMethod(PackagedProgram.java:520)
>> at
>> org.apache.flink.client.program.PackagedProgram.invokeInteractiveModeForExecution(PackagedProgram.java:403)
>> at
>> org.apache.flink.client.program.OptimizerPlanEnvironment.getOptimizedPlan(OptimizerPlanEnvironment.java:80)
>> at
>> org.apache.flink.client.program.Client.getOptimizedPlan(Client.java:215)
>> at
>> org.apache.flink.runtime.webmonitor.handlers.JarActionHandler.getJobGraphAndClassLoader(JarActionHandler.java:95)
>> at
>> org.apache.flink.runtime.webmonitor.handlers.JarPlanHandler.handleRequest(JarPlanHandler.java:42)
>> at
>> org.apache.flink.runtime.webmonitor.RuntimeMonitorHandler.respondAsLeader(RuntimeMonitorHandler.java:135)
>> at
>> org.apache.flink.runtime.webmonitor.RuntimeMonitorHandler.channelRead0(RuntimeMonitorHandler.java:112)
>> at
>> org.apache.flink.runtime.webmonitor.RuntimeMonitorHandler.channelRead0(RuntimeMonitorHandler.java:60)
>> at
>> io.netty.channel.SimpleChannelInboundHandler.channelRead(SimpleChannelInboundHandler.java:105)
>> at
>> io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:339)
>> at
>> io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:324)
>> at io.netty.handler.codec.http.router.Handler.routed(Handler.java:62)
>> at
>> io.netty.handler.codec.http.router.DualAbstractHandler.channelRead0(DualAbstractHandler.java:57)
>> at
>> io.netty.handler.codec.http.router.DualAbstractHandler.channelRead0(DualAbstractHandler.java:20)
>> at
>> io.netty.channel.SimpleChannelInboundHandler.channelRead(SimpleChannelInboundHandler.java:105)
>> at
>> io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:339)
>> at
>> io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:324)
>> at
>> org.apache.flink.runtime.webmonitor.HttpRequestHandler.channelRead0(HttpRequestHandler.java:104)
>> at
>> org.apache.flink.runtime.webmonitor.HttpRequestHandler.channelRead0(HttpRequestHandler.java:65)
>> at
>> io.netty.channel.SimpleChannelInboundHandler.channelRead(SimpleChannelInboundHandler.java:105)
>> at
>> io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:339)
>> at
>> io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:324)
>> at
>> io.netty.handler.codec.ByteToMessageDecoder.channelRead(ByteToMessageDecoder.java:242)
>> at
>> io.netty.channel.CombinedChannelDuplexHandler.channelRead(CombinedChannelDuplexHandler.java:147)
>> at
>> io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:339)
>> at
>> io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:324)
>> at
>> io.netty.channel.DefaultChannelPipeline.fireChannelRead(DefaultChannelPipeline.java:847)
>> at
>> io.netty.channel.nio.AbstractNioByteChannel$NioByteUnsafe.read(AbstractNioByteChannel.java:131)
>> at
>> io.netty.channel.nio.NioEventLoop.processSelectedKey(NioEventLoop.java:511)
>> at
>> io.netty.channel.nio.NioEventLoop.processSelectedKeysOptimized(NioEventLoop.java:468)
>> at
>> io.netty.channel.nio.NioEventLoop.processSelectedKeys(NioEventLoop.java:382)
>> at io.netty.channel.nio.NioEventLoop.run(NioEventLoop.java:354)
>> at
>> io.netty.util.concurrent.SingleThreadEventExecutor$2.run(SingleThreadEventExecutor.java:111)
>> at
>> io.netty.util.concurrent.DefaultThreadFactory$DefaultRunnableDecorator.run(DefaultThreadFactory.java:137)
>> at java.lang.Thread.run(Thread.java:745)
>> Caused by: java.lang.IllegalArgumentException: interface
>> org.apache.beam.samples.EventsByLocation$Options is not visible from class
>> loader
>> at java.lang.reflect.Proxy$ProxyClassFactory.apply(Proxy.java:581)
>> at java.lang.reflect.Proxy$ProxyClassFactory.apply(Proxy.java:557)
>> at java.lang.reflect.WeakCache$Factory.get(WeakCache.java:230)
>> at java.lang.reflect.WeakCache.get(WeakCache.java:127)
>> at java.lang.reflect.Proxy.getProxyClass0(Proxy.java:419)
>> at java.lang.reflect.Proxy.getProxyClass(Proxy.java:371)
>> at
>> org.apache.beam.sdk.options.PipelineOptionsFactory.validateWellFormed(PipelineOptionsFactory.java:620)
>> at
>> org.apache.beam.sdk.options.PipelineOptionsFactory.parseObjects(PipelineOptionsFactory.java:1374)
>> at
>> org.apache.beam.sdk.options.PipelineOptionsFactory.access$400(PipelineOptionsFactory.java:107)
>> at
>> org.apache.beam.sdk.options.PipelineOptionsFactory$Builder.as(PipelineOptionsFactory.java:292)
>> at
>> org.apache.beam.samples.EventsByLocation.main(EventsByLocation.java:62)
>> at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
>> at
>> sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
>> at
>> sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
>> at java.lang.reflect.Method.invoke(Method.java:498)
>> at
>> org.apache.flink.client.program.PackagedProgram.callMainMethod(PackagedProgram.java:505)
>> ... 35 more
>>
>> Sorry for the long exception in the message but I wanted to show all
>> possible details so you can help me fix this. Notice that the interface
>> org.apache.beam.samples.EventsByLocation$Options is a classical Beam
>> Options interface created to configure the Pipeline, and I tried making it
>> public but access does not seem to be the case, but classloading.
>>
>> Thanks,
>> Ismael
>>
>> ps.
>> I saw this and I think maybe it could be related:
>>
>> https://stackoverflow.com/questions/211176/interface-is-not-visible-from-classloader-when-using-a-proxy
>>
>>
>> On Thu, Jul 7, 2016 at 11:30 PM, Ismaël Mejía <ie...@gmail.com> wrote:
>>
>>> Both Max and Alojscha, thanks for writing, I have been a bit busy these
>>> last
>>> days with other Beam stuff and I have not been able to continue testing
>>> my Beam
>>> on flink integration, I will write back to you as soon as I have time
>>> the check
>>> this again, thanks.
>>>
>>>
>>
>
Re: A quick demo of Apache Beam with Docker
Posted by Lukasz Cwik <lc...@google.com>.
I believe I had run into this before and it was because the outer class was
not public so to make sure, you
made org.apache.beam.samples.EventsByLocation public?
Have you tried moving the org.apache.beam.samples.EventsByLocation$Options
inner class to its own file?
On Wed, Jul 13, 2016 at 4:43 AM, Ismaël Mejía <ie...@gmail.com> wrote:
> For context:
>
> I just changed the beam-runners-flink_2.10 pom file to add the shade
> plugin to
> pack the flink runner dependencies in an uber jar, and exclude the flink
> classes
> (that I assume are present on flink. My goal is to create a fat jar and
> add it to
> $FLINK_HOME/lib so I can deploy any pipeline on flink (with Beam included):
>
> <plugin>
> <groupId>org.apache.maven.plugins</groupId>
> <artifactId>maven-shade-plugin</artifactId>
> <version>2.4.1</version>
> <executions>
> <execution>
> <phase>package</phase>
> <goals>
> <goal>shade</goal>
> </goals>
> <configuration>
> <artifactSet>
> <excludes>
> <exclude>org.apache.flink:force-shading</exclude>
> <exclude>org.apache.flink:flink-clients_2.10</exclude>
> <exclude>org.apache.flink:flink-optimizer_2.10</exclude>
> <exclude>org.apache.flink:flink-java</exclude>
> <exclude>org.apache.flink:flink-shaded-hadoop2</exclude>
>
> <exclude>org.apache.flink:flink-streaming-java_2.10</exclude>
> <exclude>org.apache.flink:flink-core</exclude>
> <exclude>org.apache.flink:flink-annotations</exclude>
> <exclude>org.apache.flink:flink-runtime_2.10</exclude>
> </excludes>
> </artifactSet>
> </configuration>
> </execution>
> </executions>
>
> </plugin>
>
>
> Then I build the beam-flink-runner uber jar:
>
> mvn clean package -DskipTests
>
> And I copy the produced jar into the $FLINK_HOME/lib directory, however
> when I try to submit a
> simple test jar, via the Flink Web UI I get this exception:
>
> org.apache.flink.client.program.ProgramInvocationException: The main
> method caused an error.
> at
> org.apache.flink.client.program.PackagedProgram.callMainMethod(PackagedProgram.java:520)
> at
> org.apache.flink.client.program.PackagedProgram.invokeInteractiveModeForExecution(PackagedProgram.java:403)
> at
> org.apache.flink.client.program.OptimizerPlanEnvironment.getOptimizedPlan(OptimizerPlanEnvironment.java:80)
> at
> org.apache.flink.client.program.Client.getOptimizedPlan(Client.java:215)
> at
> org.apache.flink.runtime.webmonitor.handlers.JarActionHandler.getJobGraphAndClassLoader(JarActionHandler.java:95)
> at
> org.apache.flink.runtime.webmonitor.handlers.JarPlanHandler.handleRequest(JarPlanHandler.java:42)
> at
> org.apache.flink.runtime.webmonitor.RuntimeMonitorHandler.respondAsLeader(RuntimeMonitorHandler.java:135)
> at
> org.apache.flink.runtime.webmonitor.RuntimeMonitorHandler.channelRead0(RuntimeMonitorHandler.java:112)
> at
> org.apache.flink.runtime.webmonitor.RuntimeMonitorHandler.channelRead0(RuntimeMonitorHandler.java:60)
> at
> io.netty.channel.SimpleChannelInboundHandler.channelRead(SimpleChannelInboundHandler.java:105)
> at
> io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:339)
> at
> io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:324)
> at io.netty.handler.codec.http.router.Handler.routed(Handler.java:62)
> at
> io.netty.handler.codec.http.router.DualAbstractHandler.channelRead0(DualAbstractHandler.java:57)
> at
> io.netty.handler.codec.http.router.DualAbstractHandler.channelRead0(DualAbstractHandler.java:20)
> at
> io.netty.channel.SimpleChannelInboundHandler.channelRead(SimpleChannelInboundHandler.java:105)
> at
> io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:339)
> at
> io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:324)
> at
> org.apache.flink.runtime.webmonitor.HttpRequestHandler.channelRead0(HttpRequestHandler.java:104)
> at
> org.apache.flink.runtime.webmonitor.HttpRequestHandler.channelRead0(HttpRequestHandler.java:65)
> at
> io.netty.channel.SimpleChannelInboundHandler.channelRead(SimpleChannelInboundHandler.java:105)
> at
> io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:339)
> at
> io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:324)
> at
> io.netty.handler.codec.ByteToMessageDecoder.channelRead(ByteToMessageDecoder.java:242)
> at
> io.netty.channel.CombinedChannelDuplexHandler.channelRead(CombinedChannelDuplexHandler.java:147)
> at
> io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:339)
> at
> io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:324)
> at
> io.netty.channel.DefaultChannelPipeline.fireChannelRead(DefaultChannelPipeline.java:847)
> at
> io.netty.channel.nio.AbstractNioByteChannel$NioByteUnsafe.read(AbstractNioByteChannel.java:131)
> at
> io.netty.channel.nio.NioEventLoop.processSelectedKey(NioEventLoop.java:511)
> at
> io.netty.channel.nio.NioEventLoop.processSelectedKeysOptimized(NioEventLoop.java:468)
> at
> io.netty.channel.nio.NioEventLoop.processSelectedKeys(NioEventLoop.java:382)
> at io.netty.channel.nio.NioEventLoop.run(NioEventLoop.java:354)
> at
> io.netty.util.concurrent.SingleThreadEventExecutor$2.run(SingleThreadEventExecutor.java:111)
> at
> io.netty.util.concurrent.DefaultThreadFactory$DefaultRunnableDecorator.run(DefaultThreadFactory.java:137)
> at java.lang.Thread.run(Thread.java:745)
> Caused by: java.lang.IllegalArgumentException: interface
> org.apache.beam.samples.EventsByLocation$Options is not visible from class
> loader
> at java.lang.reflect.Proxy$ProxyClassFactory.apply(Proxy.java:581)
> at java.lang.reflect.Proxy$ProxyClassFactory.apply(Proxy.java:557)
> at java.lang.reflect.WeakCache$Factory.get(WeakCache.java:230)
> at java.lang.reflect.WeakCache.get(WeakCache.java:127)
> at java.lang.reflect.Proxy.getProxyClass0(Proxy.java:419)
> at java.lang.reflect.Proxy.getProxyClass(Proxy.java:371)
> at
> org.apache.beam.sdk.options.PipelineOptionsFactory.validateWellFormed(PipelineOptionsFactory.java:620)
> at
> org.apache.beam.sdk.options.PipelineOptionsFactory.parseObjects(PipelineOptionsFactory.java:1374)
> at
> org.apache.beam.sdk.options.PipelineOptionsFactory.access$400(PipelineOptionsFactory.java:107)
> at
> org.apache.beam.sdk.options.PipelineOptionsFactory$Builder.as(PipelineOptionsFactory.java:292)
> at
> org.apache.beam.samples.EventsByLocation.main(EventsByLocation.java:62)
> at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
> at
> sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
> at
> sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
> at java.lang.reflect.Method.invoke(Method.java:498)
> at
> org.apache.flink.client.program.PackagedProgram.callMainMethod(PackagedProgram.java:505)
> ... 35 more
>
> Sorry for the long exception in the message but I wanted to show all
> possible details so you can help me fix this. Notice that the interface
> org.apache.beam.samples.EventsByLocation$Options is a classical Beam
> Options interface created to configure the Pipeline, and I tried making it
> public but access does not seem to be the case, but classloading.
>
> Thanks,
> Ismael
>
> ps.
> I saw this and I think maybe it could be related:
>
> https://stackoverflow.com/questions/211176/interface-is-not-visible-from-classloader-when-using-a-proxy
>
>
> On Thu, Jul 7, 2016 at 11:30 PM, Ismaël Mejía <ie...@gmail.com> wrote:
>
>> Both Max and Alojscha, thanks for writing, I have been a bit busy these
>> last
>> days with other Beam stuff and I have not been able to continue testing
>> my Beam
>> on flink integration, I will write back to you as soon as I have time the
>> check
>> this again, thanks.
>>
>>
>
Re: A quick demo of Apache Beam with Docker
Posted by Ismaël Mejía <ie...@gmail.com>.
For context:
I just changed the beam-runners-flink_2.10 pom file to add the shade plugin
to
pack the flink runner dependencies in an uber jar, and exclude the flink
classes
(that I assume are present on flink. My goal is to create a fat jar and add
it to
$FLINK_HOME/lib so I can deploy any pipeline on flink (with Beam included):
<plugin>
<groupId>org.apache.maven.plugins</groupId>
<artifactId>maven-shade-plugin</artifactId>
<version>2.4.1</version>
<executions>
<execution>
<phase>package</phase>
<goals>
<goal>shade</goal>
</goals>
<configuration>
<artifactSet>
<excludes>
<exclude>org.apache.flink:force-shading</exclude>
<exclude>org.apache.flink:flink-clients_2.10</exclude>
<exclude>org.apache.flink:flink-optimizer_2.10</exclude>
<exclude>org.apache.flink:flink-java</exclude>
<exclude>org.apache.flink:flink-shaded-hadoop2</exclude>
<exclude>org.apache.flink:flink-streaming-java_2.10</exclude>
<exclude>org.apache.flink:flink-core</exclude>
<exclude>org.apache.flink:flink-annotations</exclude>
<exclude>org.apache.flink:flink-runtime_2.10</exclude>
</excludes>
</artifactSet>
</configuration>
</execution>
</executions>
</plugin>
Then I build the beam-flink-runner uber jar:
mvn clean package -DskipTests
And I copy the produced jar into the $FLINK_HOME/lib directory, however
when I try to submit a
simple test jar, via the Flink Web UI I get this exception:
org.apache.flink.client.program.ProgramInvocationException: The main method
caused an error.
at
org.apache.flink.client.program.PackagedProgram.callMainMethod(PackagedProgram.java:520)
at
org.apache.flink.client.program.PackagedProgram.invokeInteractiveModeForExecution(PackagedProgram.java:403)
at
org.apache.flink.client.program.OptimizerPlanEnvironment.getOptimizedPlan(OptimizerPlanEnvironment.java:80)
at
org.apache.flink.client.program.Client.getOptimizedPlan(Client.java:215)
at
org.apache.flink.runtime.webmonitor.handlers.JarActionHandler.getJobGraphAndClassLoader(JarActionHandler.java:95)
at
org.apache.flink.runtime.webmonitor.handlers.JarPlanHandler.handleRequest(JarPlanHandler.java:42)
at
org.apache.flink.runtime.webmonitor.RuntimeMonitorHandler.respondAsLeader(RuntimeMonitorHandler.java:135)
at
org.apache.flink.runtime.webmonitor.RuntimeMonitorHandler.channelRead0(RuntimeMonitorHandler.java:112)
at
org.apache.flink.runtime.webmonitor.RuntimeMonitorHandler.channelRead0(RuntimeMonitorHandler.java:60)
at
io.netty.channel.SimpleChannelInboundHandler.channelRead(SimpleChannelInboundHandler.java:105)
at
io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:339)
at
io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:324)
at io.netty.handler.codec.http.router.Handler.routed(Handler.java:62)
at
io.netty.handler.codec.http.router.DualAbstractHandler.channelRead0(DualAbstractHandler.java:57)
at
io.netty.handler.codec.http.router.DualAbstractHandler.channelRead0(DualAbstractHandler.java:20)
at
io.netty.channel.SimpleChannelInboundHandler.channelRead(SimpleChannelInboundHandler.java:105)
at
io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:339)
at
io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:324)
at
org.apache.flink.runtime.webmonitor.HttpRequestHandler.channelRead0(HttpRequestHandler.java:104)
at
org.apache.flink.runtime.webmonitor.HttpRequestHandler.channelRead0(HttpRequestHandler.java:65)
at
io.netty.channel.SimpleChannelInboundHandler.channelRead(SimpleChannelInboundHandler.java:105)
at
io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:339)
at
io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:324)
at
io.netty.handler.codec.ByteToMessageDecoder.channelRead(ByteToMessageDecoder.java:242)
at
io.netty.channel.CombinedChannelDuplexHandler.channelRead(CombinedChannelDuplexHandler.java:147)
at
io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:339)
at
io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:324)
at
io.netty.channel.DefaultChannelPipeline.fireChannelRead(DefaultChannelPipeline.java:847)
at
io.netty.channel.nio.AbstractNioByteChannel$NioByteUnsafe.read(AbstractNioByteChannel.java:131)
at
io.netty.channel.nio.NioEventLoop.processSelectedKey(NioEventLoop.java:511)
at
io.netty.channel.nio.NioEventLoop.processSelectedKeysOptimized(NioEventLoop.java:468)
at
io.netty.channel.nio.NioEventLoop.processSelectedKeys(NioEventLoop.java:382)
at io.netty.channel.nio.NioEventLoop.run(NioEventLoop.java:354)
at
io.netty.util.concurrent.SingleThreadEventExecutor$2.run(SingleThreadEventExecutor.java:111)
at
io.netty.util.concurrent.DefaultThreadFactory$DefaultRunnableDecorator.run(DefaultThreadFactory.java:137)
at java.lang.Thread.run(Thread.java:745)
Caused by: java.lang.IllegalArgumentException: interface
org.apache.beam.samples.EventsByLocation$Options is not visible from class
loader
at java.lang.reflect.Proxy$ProxyClassFactory.apply(Proxy.java:581)
at java.lang.reflect.Proxy$ProxyClassFactory.apply(Proxy.java:557)
at java.lang.reflect.WeakCache$Factory.get(WeakCache.java:230)
at java.lang.reflect.WeakCache.get(WeakCache.java:127)
at java.lang.reflect.Proxy.getProxyClass0(Proxy.java:419)
at java.lang.reflect.Proxy.getProxyClass(Proxy.java:371)
at
org.apache.beam.sdk.options.PipelineOptionsFactory.validateWellFormed(PipelineOptionsFactory.java:620)
at
org.apache.beam.sdk.options.PipelineOptionsFactory.parseObjects(PipelineOptionsFactory.java:1374)
at
org.apache.beam.sdk.options.PipelineOptionsFactory.access$400(PipelineOptionsFactory.java:107)
at
org.apache.beam.sdk.options.PipelineOptionsFactory$Builder.as(PipelineOptionsFactory.java:292)
at
org.apache.beam.samples.EventsByLocation.main(EventsByLocation.java:62)
at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
at
sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
at
sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
at java.lang.reflect.Method.invoke(Method.java:498)
at
org.apache.flink.client.program.PackagedProgram.callMainMethod(PackagedProgram.java:505)
... 35 more
Sorry for the long exception in the message but I wanted to show all
possible details so you can help me fix this. Notice that the interface
org.apache.beam.samples.EventsByLocation$Options is a classical Beam
Options interface created to configure the Pipeline, and I tried making it
public but access does not seem to be the case, but classloading.
Thanks,
Ismael
ps.
I saw this and I think maybe it could be related:
https://stackoverflow.com/questions/211176/interface-is-not-visible-from-classloader-when-using-a-proxy
On Thu, Jul 7, 2016 at 11:30 PM, Ismaël Mejía <ie...@gmail.com> wrote:
> Both Max and Alojscha, thanks for writing, I have been a bit busy these
> last
> days with other Beam stuff and I have not been able to continue testing my
> Beam
> on flink integration, I will write back to you as soon as I have time the
> check
> this again, thanks.
>
>
Re: A quick demo of Apache Beam with Docker
Posted by Ismaël Mejía <ie...@gmail.com>.
Both Max and Alojscha, thanks for writing, I have been a bit busy these
last
days with other Beam stuff and I have not been able to continue testing my
Beam
on flink integration, I will write back to you as soon as I have time the
check
this again, thanks.
Re: A quick demo of Apache Beam with Docker
Posted by Maximilian Michels <mx...@apache.org>.
Hi Ismael,
I'm heading out for vacations but just some thoughts on your questions:
1. My current approach is based on Emanuele’s idea to create an uber jar
> with the Beam SDK + the Flink Runner + all the Beam IOs and their
> dependencies (I exclude all org.apache.flink because those are provided by
> Flink). I put this big jar on $FLINK_HOME/lib and I start Flink. However I
> created a small Beam example jar and submitted it into Flink but I am
> having classpath issues. Do you have any suggestions, is there a better way
> to do this ? I suppose my approach is far from the best but I don’t know
> how Flink deals with this 'extension' cases.
>
Not everything in org.apache.flink is provided by Flink. The README example
POM in Beam used to filter out everything but, in fact, all connectors
(Kafka, Cassandra, Flume, etc.) are not part of the Apache Flink fat jar.
So either put these in the /lib folder or simply bundle them in the user
jar with the Shade or Assembly plugin.
> 2. I only found a way to run both the Flink’s JobManager and TaskManager
> in daemon mode. Is there an easy way to run both as normal processes? I ask
> this because the current docker image uses supervisor to keep the processes
> alive, but if we can get rid of supervisor the image will be reduced in 40
> more MB, and be really minimalistic, any ideas?
There is no command-line option yet (we should add one). You will have to
slightly modify the startup script. In "flink-daemon.sh", search for
"$JAVA_RUN" and delete the "&" from the end of the line. After the change,
Flink processes won't run daemonized anymore.
Cheers,
Max
On Fri, Jul 1, 2016 at 4:42 PM, Aljoscha Krettek <al...@apache.org>
wrote:
> Hi Ismael,
> what are the classpath issues that you are observing?
>
> Cheers,
> Aljoscha
>
> On Tue, 28 Jun 2016 at 18:00 Emanuele Cesena <em...@shopkick.com>
> wrote:
>
>> Hi Ismael,
>>
>> This is great!
>>
>> I’ll give it a try asap (last week of Q2… pretty busy :/).
>>
>> For your second question, a friend of mine was working on something
>> similar:
>>
>> https://github.com/psmiraglia/docker-flink-formula/blob/master/start-master.sh
>>
>> Best,
>>
>>
>> > On Jun 28, 2016, at 7:27 AM, Ismaël Mejía <ie...@gmail.com> wrote:
>> >
>> > Hello,
>> >
>> > Emanuele, thanks for posting your example. Congratulations, you just
>> had an amazing idea. First to help people to start with Beam, but also to
>> provide an easy way to run/test Beam pipelines on the Flink runner. This
>> second idea is really useful for me because I am running/testing ideas in
>> all the runners and this is a perfect way to do it.
>> >
>> > I started working to make the docker image you were based on (the one
>> on https://github.com/apache/flink/tree/master/flink-contrib/docker-flink
>> smaller. I just created FLINK-4118 and a PR that reduces the default image
>> in 460 MB.
>> > https://github.com/apache/flink/pull/2176 I hope the Flink guys accept
>> the changes.
>> >
>> > For anyone interested the final flink image is also available from my
>> docker account
>> >
>> > docker pull iemejia/flink
>> >
>> > I also started a project to contrib the integration of this smaller
>> version of the Flink image with Beam into Apache Beam, this probably goes
>> in the same line of work of the previous email from Max.I took the freedom
>> to rebase Emanuele changes into a big commit, and start working from there
>> https://github.com/iemejia/incubator-beam/tree/docker-flink. I hope we
>> can share our work there, of course with the people interested (e.g.
>> Emanuele and Maximilian).
>> >
>> > Max, I have two questions:
>> >
>> > 1. My current approach is based on Emanuele’s idea to create an uber
>> jar with the Beam SDK + the Flink Runner + all the Beam IOs and their
>> dependencies (I exclude all org.apache.flink because those are provided by
>> Flink). I put this big jar on $FLINK_HOME/lib and I start Flink. However I
>> created a small Beam example jar and submitted it into Flink but I am
>> having classpath issues. Do you have any suggestions, is there a better way
>> to do this ? I suppose my approach is far from the best but I don’t know
>> how Flink deals with this 'extension' cases.
>> >
>> > 2. I only found a way to run both the Flink’s JobManager and
>> TaskManager in daemon mode. Is there an easy way to run both as normal
>> processes? I ask this because the current docker image uses supervisor to
>> keep the processes alive, but if we can get rid of supervisor the image
>> will be reduced in 40 more MB, and be really minimalistic, any ideas?
>> >
>> > Regards,
>> > Ismael
>> >
>> > ps. Amit and JB, if you want I can prepare a docker image for the spark
>> runner, probably using the spark-job-server image as a base, I still have
>> to check how viable is this but I think is feasible.
>> >
>> >
>> >
>> > On Tue, Jun 28, 2016 at 1:39 PM, Maximilian Michels <mx...@apache.org>
>> wrote:
>> > Thanks for sharing Emanuele! Looking forward to providing built-in
>> > Docker support in Beam.
>> >
>> > On Fri, Jun 24, 2016 at 9:30 AM, Amit Sela <am...@gmail.com>
>> wrote:
>> > > You're right about standalone, I know many companies (small-medium)
>> > > companies that prefer spawning standalone per use case/s. I'm
>> currently
>> > > biased now towards large clusters because of my current work place ;)
>> which
>> > > relates better to my previous comment.
>> > >
>> > >
>> > > On Fri, Jun 24, 2016, 03:42 Emanuele Cesena <em...@shopkick.com>
>> wrote:
>> > >>
>> > >> Thanks Amit!
>> > >>
>> > >> I chose Flink because of the current capability support and for the
>> nicer
>> > >> front end UI, but I have nothing against Spark — actually I’m using
>> Spark in
>> > >> my daily job, and chances are that if we’ll use Beam, it will be on
>> Spark
>> > >> first.
>> > >>
>> > >> I can also tell you that I know of 2 instances (MemSQL, that
>> distribute
>> > >> its own Spark, and our parent company SK Planet in Korea) that
>> prefer Spark
>> > >> standalone, mostly for performance and easy of setup. So I can see a
>> lot of
>> > >> potential even in production environments.
>> > >>
>> > >> Best,
>> > >>
>> > >>
>> > >> > On Jun 23, 2016, at 3:42 PM, Amit Sela <am...@gmail.com>
>> wrote:
>> > >> >
>> > >> > Thanks for sharing Emanuele, I will definitely look into trying
>> > >> > something like that with Spark as well :)
>> > >> > While production clusters (usually) use YARN/Mesos to manage
>> resources,
>> > >> > this could be really great for developers to use on a virtual
>> environment.
>> > >> > Really interesting!
>> > >> >
>> > >> > On Thu, Jun 23, 2016 at 7:21 PM Emanuele Cesena <
>> emanuele@shopkick.com>
>> > >> > wrote:
>> > >> > Thank you Aljoscha!
>> > >> >
>> > >> > > On Jun 23, 2016, at 1:19 AM, Aljoscha Krettek <
>> aljoscha@apache.org>
>> > >> > > wrote:
>> > >> > >
>> > >> > > It's a very nice write up indeed! Thanks for sharing. :-)
>> > >> > >
>> > >> > > On Thu, 23 Jun 2016 at 07:35 Jean-Baptiste Onofré <
>> jb@nanthrax.net>
>> > >> > > wrote:
>> > >> > > Hi Emanuele,
>> > >> > >
>> > >> > > this is a great example !
>> > >> > >
>> > >> > > It shows Beam with Flink. Maybe we can enhance a bit showing how
>> the
>> > >> > > same pipeline can result to different docker depending of the
>> backend.
>> > >> > >
>> > >> > > I'm working on new "concrete" Beam samples showing that:
>> > >> > >
>> > >> > > https://github.com/jbonofre/beam-samples
>> > >> > >
>> > >> > > Great work anyway !
>> > >> > >
>> > >> > > Regards
>> > >> > > JB
>> > >> > >
>> > >> > > On 06/22/2016 10:18 PM, Emanuele Cesena wrote:
>> > >> > > > Hi,
>> > >> > > >
>> > >> > > > I just published a "quick start" with Beam and wanted to share:
>> > >> > > >
>> > >> > > >
>> https://medium.com/@ecesena/a-quick-demo-of-apache-beam-with-docker-da98b99a502a
>> > >> > > >
>> > >> > > > Related repos:
>> > >> > > > https://github.com/ecesena/docker-beam-flink
>> > >> > > > https://github.com/ecesena/beam-starter
>> > >> > > >
>> > >> > > > Any feedback is more than welcome!
>> > >> > > >
>> > >> > > > Best,
>> > >> > > > E.
>> > >> > > >
>> > >> > >
>> > >> > > --
>> > >> > > Jean-Baptiste Onofré
>> > >> > > jbonofre@apache.org
>> > >> > > http://blog.nanthrax.net
>> > >> > > Talend - http://www.talend.com
>> > >> >
>> > >>
>> > >
>> >
>>
>>