You are viewing a plain text version of this content. The canonical link for it is here.
Posted to dev@beam.apache.org by Emanuele Cesena <em...@shopkick.com> on 2016/07/24 22:33:13 UTC
Flink runner and KafkaIO
Hi,
I’ve tried creating a small stream word count with FlinkRunner and KafkaIO.
I get the exception below.
Is this related to:
https://issues.apache.org/jira/browse/BEAM-102
(in which case I’ll wait) or am I doing something wrong?
Thank you!
Code is here:
https://github.com/ecesena/beam-starter/blob/master/src/main/java/com/dataradiant/beam/examples/StreamWordCount.java#L138
Exception details:
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:744)
Caused by: java.lang.IllegalStateException: Unable to create a side-input view from input
at org.apache.beam.sdk.transforms.View$AsIterable.validate(View.java:276)
at org.apache.beam.sdk.transforms.View$AsIterable.validate(View.java:267)
at org.apache.beam.sdk.Pipeline.applyInternal(Pipeline.java:400)
at org.apache.beam.sdk.Pipeline.applyTransform(Pipeline.java:308)
at org.apache.beam.sdk.values.PCollection.apply(PCollection.java:154)
at org.apache.beam.sdk.io.Write$Bound.createWrite(Write.java:381)
at org.apache.beam.sdk.io.Write$Bound.apply(Write.java:111)
at org.apache.beam.sdk.io.Write$Bound.apply(Write.java:98)
at org.apache.beam.sdk.runners.PipelineRunner.apply(PipelineRunner.java:76)
at org.apache.beam.runners.flink.FlinkRunner.apply(FlinkRunner.java:140)
at org.apache.beam.sdk.Pipeline.applyInternal(Pipeline.java:401)
at org.apache.beam.sdk.Pipeline.applyTransform(Pipeline.java:324)
at org.apache.beam.sdk.values.PCollection.apply(PCollection.java:167)
at org.apache.beam.sdk.io.TextIO$Write$Bound.apply(TextIO.java:617)
at org.apache.beam.sdk.io.TextIO$Write$Bound.apply(TextIO.java:463)
at org.apache.beam.sdk.runners.PipelineRunner.apply(PipelineRunner.java:76)
at org.apache.beam.runners.flink.FlinkRunner.apply(FlinkRunner.java:140)
at org.apache.beam.sdk.Pipeline.applyInternal(Pipeline.java:401)
at org.apache.beam.sdk.Pipeline.applyTransform(Pipeline.java:324)
at org.apache.beam.sdk.values.PCollection.apply(PCollection.java:167)
at com.dataradiant.beam.examples.StreamWordCount.main(StreamWordCount.java:138)
at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:57)
at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
at java.lang.reflect.Method.invoke(Method.java:606)
at org.apache.flink.client.program.PackagedProgram.callMainMethod(PackagedProgram.java:505)
... 35 more
Caused by: java.lang.IllegalStateException: GroupByKey cannot be applied to non-bounded PCollection in the GlobalWindow without a trigger. Use a Window.into or Window.triggering transform prior to GroupByKey.
at org.apache.beam.sdk.transforms.GroupByKey.applicableTo(GroupByKey.java:173)
at org.apache.beam.sdk.transforms.View$AsIterable.validate(View.java:274)
... 60 more
--
Emanuele Cesena, Data Eng.
http://www.shopkick.com
Il corpo non ha ideali
Re: Flink runner and KafkaIO
Posted by Emanuele Cesena <em...@shopkick.com>.
Great, thank you!
> On Jul 24, 2016, at 11:04 PM, Aljoscha Krettek <al...@apache.org> wrote:
>
> Hi,
> the root cause seems to be that the Flink streaming runner does not support
> side-inputs (Views). Inside the CountWords it uses a Combine somewhere
> which uses side-inputs to sneak in a default value in case there is nothing
> to combine.
>
> Good news is that I have almost finished work on making side-inputs work in
> Flink. I'll let you know once that is in master.
>
> Cheers,
> Aljoscha
>
> On Mon, 25 Jul 2016 at 00:33 Emanuele Cesena <em...@shopkick.com> wrote:
>
>> Hi,
>>
>> I’ve tried creating a small stream word count with FlinkRunner and KafkaIO.
>> I get the exception below.
>>
>> Is this related to:
>> https://issues.apache.org/jira/browse/BEAM-102
>> (in which case I’ll wait) or am I doing something wrong?
>>
>> Thank you!
>>
>>
>> Code is here:
>>
>> https://github.com/ecesena/beam-starter/blob/master/src/main/java/com/dataradiant/beam/examples/StreamWordCount.java#L138
>>
>> Exception details:
>> 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:744)
>> Caused by: java.lang.IllegalStateException: Unable to create a side-input
>> view from input
>> at
>> org.apache.beam.sdk.transforms.View$AsIterable.validate(View.java:276)
>> at
>> org.apache.beam.sdk.transforms.View$AsIterable.validate(View.java:267)
>> at org.apache.beam.sdk.Pipeline.applyInternal(Pipeline.java:400)
>> at org.apache.beam.sdk.Pipeline.applyTransform(Pipeline.java:308)
>> at
>> org.apache.beam.sdk.values.PCollection.apply(PCollection.java:154)
>> at org.apache.beam.sdk.io.Write$Bound.createWrite(Write.java:381)
>> at org.apache.beam.sdk.io.Write$Bound.apply(Write.java:111)
>> at org.apache.beam.sdk.io.Write$Bound.apply(Write.java:98)
>> at
>> org.apache.beam.sdk.runners.PipelineRunner.apply(PipelineRunner.java:76)
>> at
>> org.apache.beam.runners.flink.FlinkRunner.apply(FlinkRunner.java:140)
>> at org.apache.beam.sdk.Pipeline.applyInternal(Pipeline.java:401)
>> at org.apache.beam.sdk.Pipeline.applyTransform(Pipeline.java:324)
>> at
>> org.apache.beam.sdk.values.PCollection.apply(PCollection.java:167)
>> at org.apache.beam.sdk.io.TextIO$Write$Bound.apply(TextIO.java:617)
>> at org.apache.beam.sdk.io.TextIO$Write$Bound.apply(TextIO.java:463)
>> at
>> org.apache.beam.sdk.runners.PipelineRunner.apply(PipelineRunner.java:76)
>> at
>> org.apache.beam.runners.flink.FlinkRunner.apply(FlinkRunner.java:140)
>> at org.apache.beam.sdk.Pipeline.applyInternal(Pipeline.java:401)
>> at org.apache.beam.sdk.Pipeline.applyTransform(Pipeline.java:324)
>> at
>> org.apache.beam.sdk.values.PCollection.apply(PCollection.java:167)
>> at
>> com.dataradiant.beam.examples.StreamWordCount.main(StreamWordCount.java:138)
>> at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
>> at
>> sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:57)
>> at
>> sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
>> at java.lang.reflect.Method.invoke(Method.java:606)
>> at
>> org.apache.flink.client.program.PackagedProgram.callMainMethod(PackagedProgram.java:505)
>> ... 35 more
>> Caused by: java.lang.IllegalStateException: GroupByKey cannot be applied
>> to non-bounded PCollection in the GlobalWindow without a trigger. Use a
>> Window.into or Window.triggering transform prior to GroupByKey.
>> at
>> org.apache.beam.sdk.transforms.GroupByKey.applicableTo(GroupByKey.java:173)
>> at
>> org.apache.beam.sdk.transforms.View$AsIterable.validate(View.java:274)
>> ... 60 more
>>
>>
>> --
>> Emanuele Cesena, Data Eng.
>> http://www.shopkick.com
>>
>> Il corpo non ha ideali
>>
>>
>>
>>
>>
--
Emanuele Cesena, Data Eng.
http://www.shopkick.com
Il corpo non ha ideali
Re: Flink runner and KafkaIO
Posted by Aljoscha Krettek <al...@apache.org>.
Hi,
the root cause seems to be that the Flink streaming runner does not support
side-inputs (Views). Inside the CountWords it uses a Combine somewhere
which uses side-inputs to sneak in a default value in case there is nothing
to combine.
Good news is that I have almost finished work on making side-inputs work in
Flink. I'll let you know once that is in master.
Cheers,
Aljoscha
On Mon, 25 Jul 2016 at 00:33 Emanuele Cesena <em...@shopkick.com> wrote:
> Hi,
>
> I’ve tried creating a small stream word count with FlinkRunner and KafkaIO.
> I get the exception below.
>
> Is this related to:
> https://issues.apache.org/jira/browse/BEAM-102
> (in which case I’ll wait) or am I doing something wrong?
>
> Thank you!
>
>
> Code is here:
>
> https://github.com/ecesena/beam-starter/blob/master/src/main/java/com/dataradiant/beam/examples/StreamWordCount.java#L138
>
> Exception details:
> 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:744)
> Caused by: java.lang.IllegalStateException: Unable to create a side-input
> view from input
> at
> org.apache.beam.sdk.transforms.View$AsIterable.validate(View.java:276)
> at
> org.apache.beam.sdk.transforms.View$AsIterable.validate(View.java:267)
> at org.apache.beam.sdk.Pipeline.applyInternal(Pipeline.java:400)
> at org.apache.beam.sdk.Pipeline.applyTransform(Pipeline.java:308)
> at
> org.apache.beam.sdk.values.PCollection.apply(PCollection.java:154)
> at org.apache.beam.sdk.io.Write$Bound.createWrite(Write.java:381)
> at org.apache.beam.sdk.io.Write$Bound.apply(Write.java:111)
> at org.apache.beam.sdk.io.Write$Bound.apply(Write.java:98)
> at
> org.apache.beam.sdk.runners.PipelineRunner.apply(PipelineRunner.java:76)
> at
> org.apache.beam.runners.flink.FlinkRunner.apply(FlinkRunner.java:140)
> at org.apache.beam.sdk.Pipeline.applyInternal(Pipeline.java:401)
> at org.apache.beam.sdk.Pipeline.applyTransform(Pipeline.java:324)
> at
> org.apache.beam.sdk.values.PCollection.apply(PCollection.java:167)
> at org.apache.beam.sdk.io.TextIO$Write$Bound.apply(TextIO.java:617)
> at org.apache.beam.sdk.io.TextIO$Write$Bound.apply(TextIO.java:463)
> at
> org.apache.beam.sdk.runners.PipelineRunner.apply(PipelineRunner.java:76)
> at
> org.apache.beam.runners.flink.FlinkRunner.apply(FlinkRunner.java:140)
> at org.apache.beam.sdk.Pipeline.applyInternal(Pipeline.java:401)
> at org.apache.beam.sdk.Pipeline.applyTransform(Pipeline.java:324)
> at
> org.apache.beam.sdk.values.PCollection.apply(PCollection.java:167)
> at
> com.dataradiant.beam.examples.StreamWordCount.main(StreamWordCount.java:138)
> at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
> at
> sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:57)
> at
> sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
> at java.lang.reflect.Method.invoke(Method.java:606)
> at
> org.apache.flink.client.program.PackagedProgram.callMainMethod(PackagedProgram.java:505)
> ... 35 more
> Caused by: java.lang.IllegalStateException: GroupByKey cannot be applied
> to non-bounded PCollection in the GlobalWindow without a trigger. Use a
> Window.into or Window.triggering transform prior to GroupByKey.
> at
> org.apache.beam.sdk.transforms.GroupByKey.applicableTo(GroupByKey.java:173)
> at
> org.apache.beam.sdk.transforms.View$AsIterable.validate(View.java:274)
> ... 60 more
>
>
> --
> Emanuele Cesena, Data Eng.
> http://www.shopkick.com
>
> Il corpo non ha ideali
>
>
>
>
>