We solved the problem of serialization by making some things transient
which were being captured as part of the closure. So we no longer have
serialization errors. Everything works properly without the future.
I realize that because of statics concurrent job submission will be an
issue. But we are submitting one job only - the difference is that it's
through a Future. So there is no concurrent submission unless I am missing
something.
regards.
On Thu, Sep 19, 2019 at 12:54 PM Biao Liu <mmyy1...@gmail.com> wrote:
> Hi Debasish,
>
> I think there is something critical of your usage hided. It might help if
> you could provide more details.
>
> It still confuses me how you solve the serialization issue. Why the
> non-transient fields only affects serialization in a future?
>
> WRT this ProgramAbortException issue, do you submit jobs concurrently in
> one process?
> Currently job submission is not thread-safe. It relies on some static
> variables which could be affected by other concurrent submissions in the
> same process.
> Asking this because usually job submission is not through
> OptimizerPlanEnvironment which appears in your exception stack trace.
>
> Thanks,
> Biao /'bɪ.aʊ/
>
>
>
> On Thu, 19 Sep 2019 at 15:03, Debasish Ghosh <ghosh.debas...@gmail.com>
> wrote:
>
>> I think what you are pointing at is asynchronous datastream operations.
>> In our case we want to submit the entire job in a Future. Something like
>> the following ..
>>
>> def execute(..) = {
>> // this does all data stream manipulation, joins etc.
>> buildComputationGraph()
>>
>> // submits for execution with StreamExecutionEnvironment
>> env.execute(..)
>> }
>>
>> and we want to do ..
>>
>> val jobExecutionResultFuture = Future(execute(..))
>>
>> and this gives that exception.
>>
>> regards.
>>
>> On Thu, Sep 19, 2019 at 11:00 AM Rafi Aroch <rafi.ar...@gmail.com> wrote:
>>
>>> Hi Debasish,
>>>
>>> Have you taken a look at the AsyncIO API for running async operations? I
>>> think this is the preferred way of doing it. [1]
>>> So it would look something like this:
>>>
>>> class AsyncDatabaseRequest extends AsyncFunction[String, (String, String)] {
>>>
>>> /** The database specific client that can issue concurrent requests
>>> with callbacks */
>>> lazy val client: DatabaseClient = new DatabaseClient(host, post,
>>> credentials)
>>>
>>> /** The context used for the future callbacks */
>>> implicit lazy val executor: ExecutionContext =
>>> ExecutionContext.fromExecutor(Executors.directExecutor())
>>>
>>>
>>> override def asyncInvoke(str: String, resultFuture:
>>> ResultFuture[(String, String)]): Unit = {
>>>
>>> // issue the asynchronous request, receive a future for the result
>>> val resultFutureRequested: Future[String] = client.query(str)
>>>
>>> // set the callback to be executed once the request by the client
>>> is complete // the callback simply forwards the result to the result
>>> future resultFutureRequested.onSuccess {
>>> case result: String => resultFuture.complete(Iterable((str,
>>> result)))
>>> }
>>> }}
>>>
>>> // create the original streamval stream: DataStream[String] = ...
>>> // apply the async I/O transformationval resultStream: DataStream[(String,
>>> String)] =
>>> AsyncDataStream.unorderedWait(stream, new AsyncDatabaseRequest(), 1000,
>>> TimeUnit.MILLISECONDS, 100)
>>>
>>>
>>> Thanks,
>>> Rafi
>>>
>>> [1]
>>> https://ci.apache.org/projects/flink/flink-docs-stable/dev/stream/operators/asyncio.html#async-io-api
>>>
>>> On Wed, Sep 18, 2019 at 8:26 PM Debasish Ghosh <ghosh.debas...@gmail.com>
>>> wrote:
>>>
>>>> ok, the above problem was due to some serialization issues which we
>>>> fixed by marking some of the things transient. This fixes the serialization
>>>> issues .. But now when I try to execute in a Future I hit upon this ..
>>>>
>>>>
>>>> *java.util.concurrent.ExecutionException: Boxed Error* at
>>>> scala.concurrent.impl.Promise$.resolver(Promise.scala:87)
>>>> at
>>>> scala.concurrent.impl.Promise$.scala$concurrent$impl$Promise$$resolveTry(Promise.scala:79)
>>>> at
>>>> scala.concurrent.impl.Promise$DefaultPromise.tryComplete(Promise.scala:284)
>>>> at scala.concurrent.Promise.complete(Promise.scala:53)
>>>> at scala.concurrent.Promise.complete$(Promise.scala:52)
>>>> at
>>>> scala.concurrent.impl.Promise$DefaultPromise.complete(Promise.scala:187)
>>>> at scala.concurrent.impl.Promise.$anonfun$transform$1(Promise.scala:33)
>>>> at scala.concurrent.impl.CallbackRunnable.run(Promise.scala:64)
>>>> at
>>>> java.util.concurrent.ForkJoinTask$RunnableExecuteAction.exec(ForkJoinTask.java:1402)
>>>> at java.util.concurrent.ForkJoinTask.doExec(ForkJoinTask.java:289)
>>>> at
>>>> java.util.concurrent.ForkJoinPool$WorkQueue.runTask(ForkJoinPool.java:1056)
>>>> at java.util.concurrent.ForkJoinPool.runWorker(ForkJoinPool.java:1692)
>>>> at
>>>> java.util.concurrent.ForkJoinWorkerThread.run(ForkJoinWorkerThread.java:157)
>>>>
>>>> *Caused by:
>>>> org.apache.flink.client.program.OptimizerPlanEnvironment$ProgramAbortException*
>>>> at
>>>> org.apache.flink.streaming.api.environment.StreamPlanEnvironment.execute(StreamPlanEnvironment.java:66)
>>>> at
>>>> org.apache.flink.streaming.api.environment.StreamExecutionEnvironment.execute(StreamExecutionEnvironment.java:1507)
>>>> at
>>>> org.apache.flink.streaming.api.scala.StreamExecutionEnvironment.execute(StreamExecutionEnvironment.scala:654)
>>>> at
>>>> pipelines.flink.FlinkStreamletLogic.executeStreamingQueries(FlinkStreamlet.scala:280)
>>>> at
>>>> pipelines.flink.FlinkStreamlet.$anonfun$run$2(FlinkStreamlet.scala:149)
>>>> at scala.concurrent.Future$.$anonfun$apply$1(Future.scala:658)
>>>> at scala.util.Success.$anonfun$map$1(Try.scala:255)
>>>> at scala.util.Success.map(Try.scala:213)
>>>> at scala.concurrent.Future.$anonfun$map$1(Future.scala:292)
>>>> at scala.concurrent.impl.Promise.liftedTree1$1(Promise.scala:33)
>>>> ... 7 more
>>>>
>>>> I found this issue in JIRA
>>>> https://issues.apache.org/jira/browse/FLINK-10381 which is still open
>>>> and talks about a related issue. But we are not submitting multiple jobs -
>>>> we are just submitting 1 job but async in a Future. I am not clear why this
>>>> should create the problem that I see.
>>>>
>>>> Can anyone please help with an explanation ?
>>>>
>>>> regards.
>>>>
>>>> On Wed, Sep 18, 2019 at 12:22 AM Debasish Ghosh <
>>>> ghosh.debas...@gmail.com> wrote:
>>>>
>>>>> I think the issue may not be linked with Future. What happens is when
>>>>> this piece of code is executed ..
>>>>>
>>>>> val rides: DataStream[TaxiRide] =
>>>>> readStream(inTaxiRide)
>>>>> .filter { ride ⇒ ride.getIsStart().booleanValue }
>>>>> .keyBy("rideId")
>>>>>
>>>>> val fares: DataStream[TaxiFare] =
>>>>> readStream(inTaxiFare)
>>>>> .keyBy("rideId")
>>>>>
>>>>> val processed: DataStream[TaxiRideFare] =
>>>>> rides
>>>>> .connect(fares)
>>>>> .flatMap(new EnrichmentFunction)
>>>>>
>>>>> somehow the ClosureCleaner gets executed as evident from the following
>>>>> which tries to serialize Avro data. Is there any way to pass the custom
>>>>> avro serializer that I am using ?
>>>>>
>>>>> org.apache.flink.api.common.InvalidProgramException: [rideId type:LONG
>>>>> pos:0, isStart type:BOOLEAN pos:1, taxiId type:LONG pos:2, passengerCnt
>>>>> type:INT pos:3, driverId type:LONG pos:4, startLon type:FLOAT pos:5,
>>>>> startLat type:FLOAT pos:6, endLon type:FLOAT pos:7, endLat type:FLOAT
>>>>> pos:8, startTime type:LONG pos:9, endTime type:LONG pos:10] is not
>>>>> serializable. The object probably contains or references non serializable
>>>>> fields.
>>>>> at
>>>>> org.apache.flink.api.java.ClosureCleaner.clean(ClosureCleaner.java:151)
>>>>> at
>>>>> org.apache.flink.api.java.ClosureCleaner.clean(ClosureCleaner.java:126)
>>>>> at
>>>>> org.apache.flink.api.java.ClosureCleaner.clean(ClosureCleaner.java:126)
>>>>> at
>>>>> org.apache.flink.api.java.ClosureCleaner.clean(ClosureCleaner.java:126)
>>>>> at
>>>>> org.apache.flink.api.java.ClosureCleaner.clean(ClosureCleaner.java:126)
>>>>> at
>>>>> org.apache.flink.api.java.ClosureCleaner.clean(ClosureCleaner.java:71)
>>>>> at
>>>>> org.apache.flink.streaming.api.environment.StreamExecutionEnvironment.clean(StreamExecutionEnvironment.java:1574)
>>>>> at
>>>>> org.apache.flink.streaming.api.datastream.DataStream.clean(DataStream.java:185)
>>>>> at
>>>>> org.apache.flink.streaming.api.datastream.ConnectedStreams.flatMap(ConnectedStreams.java:274)
>>>>> at
>>>>> org.apache.flink.streaming.api.scala.ConnectedStreams.flatMap(ConnectedStreams.scala:179)
>>>>> at
>>>>> pipelines.examples.processor.TaxiRideProcessor$$anon$1.buildExecutionGraph(TaxiRideProcessor.scala:47)
>>>>> at
>>>>> pipelines.flink.FlinkStreamletLogic.executeStreamingQueries(FlinkStreamlet.scala:278)
>>>>> at
>>>>> pipelines.flink.FlinkStreamlet.run(FlinkStreamlet.scala:149)
>>>>> at pipelines.runner.Runner$.$anonfun$run$3(Runner.scala:44)
>>>>> at scala.util.Try$.apply(Try.scala:213)
>>>>> at pipelines.runner.Runner$.run(Runner.scala:43)
>>>>> at pipelines.runner.Runner$.main(Runner.scala:30)
>>>>> at pipelines.runner.Runner.main(Runner.scala)
>>>>> 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:576)
>>>>> at
>>>>> org.apache.flink.client.program.PackagedProgram.invokeInteractiveModeForExecution(PackagedProgram.java:438)
>>>>> at
>>>>> org.apache.flink.client.program.OptimizerPlanEnvironment.getOptimizedPlan(OptimizerPlanEnvironment.java:83)
>>>>> at
>>>>> org.apache.flink.client.program.PackagedProgramUtils.createJobGraph(PackagedProgramUtils.java:80)
>>>>> at
>>>>> org.apache.flink.runtime.webmonitor.handlers.utils.JarHandlerUtils$JarHandlerContext.toJobGraph(JarHandlerUtils.java:126)
>>>>> at
>>>>> org.apache.flink.runtime.webmonitor.handlers.JarRunHandler.lambda$getJobGraphAsync$6(JarRunHandler.java:142)
>>>>> at
>>>>> java.util.concurrent.CompletableFuture$AsyncSupply.run(CompletableFuture.java:1590)
>>>>> at
>>>>> java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
>>>>> at
>>>>> java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
>>>>> at java.lang.Thread.run(Thread.java:748)
>>>>> Caused by: java.io.NotSerializableException:
>>>>> org.apache.avro.Schema$Field
>>>>> at
>>>>> java.io.ObjectOutputStream.writeObject0(ObjectOutputStream.java:1184)
>>>>> at
>>>>> java.io.ObjectOutputStream.writeObject(ObjectOutputStream.java:348)
>>>>> at java.util.ArrayList.writeObject(ArrayList.java:766)
>>>>> 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
>>>>> java.io.ObjectStreamClass.invokeWriteObject(ObjectStreamClass.java:1140)
>>>>> at
>>>>> java.io.ObjectOutputStream.writeSerialData(ObjectOutputStream.java:1496)
>>>>> at
>>>>> java.io.ObjectOutputStream.writeOrdinaryObject(ObjectOutputStream.java:1432)
>>>>> at
>>>>> java.io.ObjectOutputStream.writeObject0(ObjectOutputStream.java:1178)
>>>>> at
>>>>> java.io.ObjectOutputStream.writeObject(ObjectOutputStream.java:348)
>>>>> at
>>>>> org.apache.flink.util.InstantiationUtil.serializeObject(InstantiationUtil.java:586)
>>>>> at
>>>>> org.apache.flink.api.java.ClosureCleaner.clean(ClosureCleaner.java:133)
>>>>>
>>>>>
>>>>> I also tried the following ..
>>>>>
>>>>> class EnrichmentFunction extends RichCoFlatMapFunction[TaxiRide,
>>>>> TaxiFare, TaxiRideFare] {
>>>>>
>>>>> @transient var rideState: ValueState[TaxiRide] = null
>>>>> @transient var fareState: ValueState[TaxiFare] = null
>>>>>
>>>>> override def open(params: Configuration): Unit = {
>>>>> super.open(params)
>>>>> rideState = getRuntimeContext.getState(
>>>>> new ValueStateDescriptor[TaxiRide]("saved ride",
>>>>> classOf[TaxiRide]))
>>>>> fareState = getRuntimeContext.getState(
>>>>> new ValueStateDescriptor[TaxiFare]("saved fare",
>>>>> classOf[TaxiFare]))
>>>>> }
>>>>>
>>>>> and moved the state initialization to open function. But still get the
>>>>> same result.
>>>>>
>>>>> Help ?
>>>>>
>>>>> regards.
>>>>>
>>>>>
>>>>>
>>>>> On Tue, Sep 17, 2019 at 12:28 PM Biao Liu <mmyy1...@gmail.com> wrote:
>>>>>
>>>>>> Hi Debasish,
>>>>>>
>>>>>> I guess the reason is something unexpectedly involved in
>>>>>> serialization due to a reference from inner class (anonymous class or
>>>>>> lambda expression).
>>>>>> When Flink serializes this inner class instance, it would also
>>>>>> serialize all referenced objects, for example, the outer class instance.
>>>>>> If
>>>>>> the outer class is not serializable, this error would happen.
>>>>>>
>>>>>> You could have a try to move the piece of codes to a named non-inner
>>>>>> class.
>>>>>>
>>>>>> Thanks,
>>>>>> Biao /'bɪ.aʊ/
>>>>>>
>>>>>>
>>>>>>
>>>>>> On Tue, 17 Sep 2019 at 02:06, Debasish Ghosh <
>>>>>> ghosh.debas...@gmail.com> wrote:
>>>>>>
>>>>>>> My main question is why serialisation kicks in when I try to execute
>>>>>>> within a `Future` and not otherwise.
>>>>>>>
>>>>>>> regards.
>>>>>>>
>>>>>>> On Mon, 16 Sep 2019 at 4:46 PM, Debasish Ghosh <
>>>>>>> ghosh.debas...@gmail.com> wrote:
>>>>>>>
>>>>>>>> Yes, they are generated from Avro Schema and implements
>>>>>>>> Serializable ..
>>>>>>>>
>>>>>>>> On Mon, Sep 16, 2019 at 4:40 PM Deepak Sharma <
>>>>>>>> deepakmc...@gmail.com> wrote:
>>>>>>>>
>>>>>>>>> Does TaxiRide or TaxiRideFare implements Serializable?
>>>>>>>>>
>>>>>>>>> On Mon, Sep 16, 2019 at 3:47 PM Debasish Ghosh <
>>>>>>>>> ghosh.debas...@gmail.com> wrote:
>>>>>>>>>
>>>>>>>>>> Hello -
>>>>>>>>>>
>>>>>>>>>> The following piece of code is an example of a connected data
>>>>>>>>>> streams ..
>>>>>>>>>>
>>>>>>>>>> val rides: DataStream[TaxiRide] =
>>>>>>>>>> readStream(inTaxiRide)
>>>>>>>>>> .filter { ride ⇒ ride.getIsStart().booleanValue }
>>>>>>>>>> .keyBy("rideId")
>>>>>>>>>>
>>>>>>>>>> val fares: DataStream[TaxiFare] =
>>>>>>>>>> readStream(inTaxiFare)
>>>>>>>>>> .keyBy("rideId")
>>>>>>>>>>
>>>>>>>>>> val processed: DataStream[TaxiRideFare] =
>>>>>>>>>> rides
>>>>>>>>>> .connect(fares)
>>>>>>>>>> .flatMap(new EnrichmentFunction)
>>>>>>>>>>
>>>>>>>>>> When I execute the above logic using
>>>>>>>>>> StreamExecutionEnvironment.execute(..) it runs fine.
>>>>>>>>>> But if I try to execute the above from within a
>>>>>>>>>> scala.concurrent.Future, I get the following exception ..
>>>>>>>>>>
>>>>>>>>>> org.apache.flink.api.common.InvalidProgramException: [rideId
>>>>>>>>>> type:LONG pos:0, isStart type:BOOLEAN pos:1, taxiId type:LONG pos:2,
>>>>>>>>>> passengerCnt type:INT pos:3, driverId type:LONG pos:4, startLon
>>>>>>>>>> type:FLOAT
>>>>>>>>>> pos:5, startLat type:FLOAT pos:6, endLon type:FLOAT pos:7, endLat
>>>>>>>>>> type:FLOAT pos:8, startTime type:LONG pos:9, endTime type:LONG
>>>>>>>>>> pos:10] is
>>>>>>>>>> not serializable. The object probably contains or references non
>>>>>>>>>> serializable fields.
>>>>>>>>>> at
>>>>>>>>>> org.apache.flink.api.java.ClosureCleaner.clean(ClosureCleaner.java:151)
>>>>>>>>>> at
>>>>>>>>>> org.apache.flink.api.java.ClosureCleaner.clean(ClosureCleaner.java:126)
>>>>>>>>>> at
>>>>>>>>>> org.apache.flink.api.java.ClosureCleaner.clean(ClosureCleaner.java:126)
>>>>>>>>>> at
>>>>>>>>>> org.apache.flink.api.java.ClosureCleaner.clean(ClosureCleaner.java:126)
>>>>>>>>>> at
>>>>>>>>>> org.apache.flink.api.java.ClosureCleaner.clean(ClosureCleaner.java:126)
>>>>>>>>>> at
>>>>>>>>>> org.apache.flink.api.java.ClosureCleaner.clean(ClosureCleaner.java:71)
>>>>>>>>>> at
>>>>>>>>>> org.apache.flink.streaming.api.environment.StreamExecutionEnvironment.clean(StreamExecutionEnvironment.java:1574)
>>>>>>>>>> at
>>>>>>>>>> org.apache.flink.streaming.api.datastream.DataStream.clean(DataStream.java:185)
>>>>>>>>>> at
>>>>>>>>>> org.apache.flink.streaming.api.datastream.ConnectedStreams.flatMap(ConnectedStreams.java:274)
>>>>>>>>>> at
>>>>>>>>>> org.apache.flink.streaming.api.scala.ConnectedStreams.flatMap(ConnectedStreams.scala:179)
>>>>>>>>>> ...
>>>>>>>>>>
>>>>>>>>>> Caused by: java.io.NotSerializableException:
>>>>>>>>>> org.apache.avro.Schema$Field
>>>>>>>>>> at
>>>>>>>>>> java.io.ObjectOutputStream.writeObject0(ObjectOutputStream.java:1184)
>>>>>>>>>>
>>>>>>>>>> Any thoughts why this may happen ?
>>>>>>>>>>
>>>>>>>>>> regards.
>>>>>>>>>>
>>>>>>>>>> --
>>>>>>>>>> Debasish Ghosh
>>>>>>>>>> http://manning.com/ghosh2
>>>>>>>>>> http://manning.com/ghosh
>>>>>>>>>>
>>>>>>>>>> Twttr: @debasishg
>>>>>>>>>> Blog: http://debasishg.blogspot.com
>>>>>>>>>> Code: http://github.com/debasishg
>>>>>>>>>>
>>>>>>>>>
>>>>>>>>>
>>>>>>>>> --
>>>>>>>>> Thanks
>>>>>>>>> Deepak
>>>>>>>>> www.bigdatabig.com
>>>>>>>>> www.keosha.net
>>>>>>>>>
>>>>>>>>
>>>>>>>>
>>>>>>>> --
>>>>>>>> Debasish Ghosh
>>>>>>>> http://manning.com/ghosh2
>>>>>>>> http://manning.com/ghosh
>>>>>>>>
>>>>>>>> Twttr: @debasishg
>>>>>>>> Blog: http://debasishg.blogspot.com
>>>>>>>> Code: http://github.com/debasishg
>>>>>>>>
>>>>>>> --
>>>>>>> Sent from my iPhone
>>>>>>>
>>>>>>
>>>>>
>>>>> --
>>>>> Debasish Ghosh
>>>>> http://manning.com/ghosh2
>>>>> http://manning.com/ghosh
>>>>>
>>>>> Twttr: @debasishg
>>>>> Blog: http://debasishg.blogspot.com
>>>>> Code: http://github.com/debasishg
>>>>>
>>>>
>>>>
>>>> --
>>>> Debasish Ghosh
>>>> http://manning.com/ghosh2
>>>> http://manning.com/ghosh
>>>>
>>>> Twttr: @debasishg
>>>> Blog: http://debasishg.blogspot.com
>>>> Code: http://github.com/debasishg
>>>>
>>>
>>
>> --
>> Debasish Ghosh
>> http://manning.com/ghosh2
>> http://manning.com/ghosh
>>
>> Twttr: @debasishg
>> Blog: http://debasishg.blogspot.com
>> Code: http://github.com/debasishg
>>
>
--
Debasish Ghosh
http://manning.com/ghosh2
http://manning.com/ghosh
Twttr: @debasishg
Blog: http://debasishg.blogspot.com
Code: http://github.com/debasishg
