Re: Could not upload the jar files to the job manager IOException

[Ana M. Martinez]

Hi Robert,

Thanks for your answer. Do you mean the log file in (e.g.) 
flink-0.10.0/log/flink-hadoop-client-ip-172-31-10-193.log? Or you mean another 
log file?

In this log, the error message is as follows:

08:16:03,437 INFO  org.apache.flink.runtime.client.JobClient
 - Job execution complete
08:16:03,438 INFO  org.apache.flink.api.java.ExecutionEnvironment   
 - The job has 5 registered types and 0 default Kryo serializers
08:16:03,444 INFO  org.apache.flink.runtime.client.JobClientActor   
 - Received job Flink Java Job at Thu Jan 21 08:16:03 UTC 2016 
(73f8ab2fbab61fb72dc4a53fd8dcbb9f).
08:16:03,444 INFO  org.apache.flink.runtime.client.JobClientActor   
 - Could not submit job Flink Java Job at Thu Jan 21 08:16:03 UTC 2016 
(73f8ab2fbab61fb72dc4a53fd8dcbb9f), because there is no connection to a 
JobManager.
08:16:03,446 INFO  org.apache.flink.runtime.client.JobClientActor   
 - Connected to new JobManager 
akka.tcp://flink@172.31.5.123:34614/user/jobmanager.
08:16:03,446 INFO  org.apache.flink.runtime.client.JobClientActor   
 - Sending message to JobManager 
akka.tcp://flink@172.31.5.123:34614/user/jobmanager to submit job Flink Java 
Job at Thu Jan 21 08:16:03 UTC 2016 (73f8ab2fbab61fb72dc4a53fd8dcbb9f) and wait 
for progress
08:16:03,446 INFO  org.apache.flink.runtime.client.JobClientActor   
 - Upload jar files to job manager 
akka.tcp://flink@172.31.5.123:34614/user/jobmanager.
08:16:03,860 INFO  org.apache.flink.runtime.client.JobClientActor   
 - Submit job to the job manager 
akka.tcp://flink@172.31.5.123:34614/user/jobmanager.
08:16:03,860 INFO  org.apache.flink.runtime.client.JobClient
 - Job execution failed
08:16:03,860 ERROR org.apache.flink.client.CliFrontend  
 - Error while running the command.
org.apache.flink.client.program.ProgramInvocationException: The main method 
caused an error.
at 
org.apache.flink.client.program.PackagedProgram.callMainMethod(PackagedProgram.java:512)
at 
org.apache.flink.client.program.PackagedProgram.invokeInteractiveModeForExecution(PackagedProgram.java:395)
at org.apache.flink.client.program.Client.runBlocking(Client.java:252)
at 
org.apache.flink.client.CliFrontend.executeProgramBlocking(CliFrontend.java:675)
at org.apache.flink.client.CliFrontend.run(CliFrontend.java:326)
at org.apache.flink.client.CliFrontend.parseParameters(CliFrontend.java:977)
at org.apache.flink.client.CliFrontend.main(CliFrontend.java:1027)
Caused by: java.lang.reflect.UndeclaredThrowableException
at 
eu.amidst.flinklink.core.io.DataFlinkLoader.loadHeaderARFFFolder(DataFlinkLoader.java:176)
at 
eu.amidst.flinklink.core.io.DataFlinkLoader.loadHeader(DataFlinkLoader.java:137)
at 
eu.amidst.flinklink.core.io.DataFlinkLoader.access$000(DataFlinkLoader.java:43)
at 
eu.amidst.flinklink.core.io.DataFlinkLoader$DataFlinkFile.(DataFlinkLoader.java:281)
at 
eu.amidst.flinklink.core.io.DataFlinkLoader.loadDataFromFolder(DataFlinkLoader.java:80)
at 
eu.amidst.flinklink.core.io.DataFlinkLoader.loadDynamicDataFromFolder(DataFlinkLoader.java:90)
at 
eu.amidst.flinklink.examples.reviewMeeting2015.GenerateData.createDataSetsDBN(GenerateData.java:194)
at 
eu.amidst.flinklink.examples.reviewMeeting2015.GenerateData.main(GenerateData.java:208)
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:497)
at 
org.apache.flink.client.program.PackagedProgram.callMainMethod(PackagedProgram.java:497)
... 6 more
Caused by: org.apache.flink.client.program.ProgramInvocationException: The 
program execution failed: Could not upload the jar files to the job manager.
at org.apache.flink.client.program.Client.runBlocking(Client.java:370)
at org.apache.flink.client.program.Client.runBlocking(Client.java:348)
at org.apache.flink.client.program.Client.runBlocking(Client.java:315)
at 
org.apache.flink.client.program.ContextEnvironment.execute(ContextEnvironment.java:70)
at 
org.apache.flink.api.java.ExecutionEnvironment.execute(ExecutionEnvironment.java:804)
at org.apache.flink.api.java.DataSet.collect(DataSet.java:410)
at 
eu.amidst.flinklink.core.io.DataFlinkLoader.loadHeaderARFFFolder(DataFlinkLoader.java:156)
... 18 more
Caused by: org.apache.flink.runtime.client.JobSubmissionException: Could not 
upload the jar files to the job manager.
at 
org.apache.flink.runtime.client.JobClientActor$2.call(JobClientActor.java:338)
at akka.dispatch.Futures$$anonfun$future$1.apply(Future.scala:94)
at 
scala.concurrent.impl.Future$PromiseCompletingRunnable.liftedTree1$1(Future.scala:24)
at scala.concurrent.impl.Future$PromiseCompletingRunnable.run(Future.scala:24)
at akka.dispatch.TaskInvocation.run(AbstractDispatcher.scala:41)
at 
akka.dispatch.ForkJoinExecu

Re: Could not upload the jar files to the job manager IOException

[Robert Metzger]

Hi,

this is the log file of your local client submitting the job to Flink's
JobManager (master). As you can see from the log, the jar upload failed
because of this issue: "Caused by: java.net.SocketException: Connection
reset". The JobManager is the service at the other end receiving the file.
I'm pretty sure its logging why the connection has been reset. Maybe the JM
jvm failed, the disk was full, there was an network issue 

The JobManager log file is located on the machine where the jobmanager is
running.
If you are using Flink on YARN, just get the aggregated application logs,
and search for the "jobmanager.log" file.

On Thu, Jan 21, 2016 at 9:27 AM, Ana M. Martinez  wrote:

>
> Hi Robert,
>
> Thanks for your answer. Do you mean the log file in (e.g.)
> flink-0.10.0/log/flink-hadoop-client-ip-172-31-10-193.log? Or you mean
> another log file?
>
> In this log, the error message is as follows:
>
> 08:16:03,437 INFO  org.apache.flink.runtime.client.JobClient
>   - Job execution complete
> 08:16:03,438 INFO  org.apache.flink.api.java.ExecutionEnvironment
>   - The job has 5 registered types and 0 default Kryo serializers
> 08:16:03,444 INFO  org.apache.flink.runtime.client.JobClientActor
>   - Received job Flink Java Job at Thu Jan 21 08:16:03 UTC 2016
> (73f8ab2fbab61fb72dc4a53fd8dcbb9f).
> 08:16:03,444 INFO  org.apache.flink.runtime.client.JobClientActor
>   - Could not submit job Flink Java Job at Thu Jan 21 08:16:03 UTC 2016
> (73f8ab2fbab61fb72dc4a53fd8dcbb9f), because there is no connection to a
> JobManager.
> 08:16:03,446 INFO  org.apache.flink.runtime.client.JobClientActor
>   - Connected to new JobManager
> akka.tcp://flink@172.31.5.123:34614/user/jobmanager.
> 08:16:03,446 INFO  org.apache.flink.runtime.client.JobClientActor
>   - Sending message to JobManager
> akka.tcp://flink@172.31.5.123:34614/user/jobmanager to submit job Flink
> Java Job at Thu Jan 21 08:16:03 UTC 2016 (73f8ab2fbab61fb72dc4a53fd8dcbb9f)
> and wait for progress
> 08:16:03,446 INFO  org.apache.flink.runtime.client.JobClientActor
>   - Upload jar files to job manager
> akka.tcp://flink@172.31.5.123:34614/user/jobmanager.
> 08:16:03,860 INFO  org.apache.flink.runtime.client.JobClientActor
>   - Submit job to the job manager
> akka.tcp://flink@172.31.5.123:34614/user/jobmanager.
> 08:16:03,860 INFO  org.apache.flink.runtime.client.JobClient
>   - Job execution failed
> 08:16:03,860 ERROR org.apache.flink.client.CliFrontend
>   - Error while running the command.
> org.apache.flink.client.program.ProgramInvocationException: The main
> method caused an error.
> at
> org.apache.flink.client.program.PackagedProgram.callMainMethod(PackagedProgram.java:512)
> at
> org.apache.flink.client.program.PackagedProgram.invokeInteractiveModeForExecution(PackagedProgram.java:395)
> at org.apache.flink.client.program.Client.runBlocking(Client.java:252)
> at
> org.apache.flink.client.CliFrontend.executeProgramBlocking(CliFrontend.java:675)
> at org.apache.flink.client.CliFrontend.run(CliFrontend.java:326)
> at
> org.apache.flink.client.CliFrontend.parseParameters(CliFrontend.java:977)
> at org.apache.flink.client.CliFrontend.main(CliFrontend.java:1027)
> Caused by: java.lang.reflect.UndeclaredThrowableException
> at
> eu.amidst.flinklink.core.io.DataFlinkLoader.loadHeaderARFFFolder(DataFlinkLoader.java:176)
> at
> eu.amidst.flinklink.core.io.DataFlinkLoader.loadHeader(DataFlinkLoader.java:137)
> at
> eu.amidst.flinklink.core.io.DataFlinkLoader.access$000(DataFlinkLoader.java:43)
> at
> eu.amidst.flinklink.core.io.DataFlinkLoader$DataFlinkFile.(DataFlinkLoader.java:281)
> at
> eu.amidst.flinklink.core.io.DataFlinkLoader.loadDataFromFolder(DataFlinkLoader.java:80)
> at
> eu.amidst.flinklink.core.io.DataFlinkLoader.loadDynamicDataFromFolder(DataFlinkLoader.java:90)
> at
> eu.amidst.flinklink.examples.reviewMeeting2015.GenerateData.createDataSetsDBN(GenerateData.java:194)
> at
> eu.amidst.flinklink.examples.reviewMeeting2015.GenerateData.main(GenerateData.java:208)
> 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:497)
> at
> org.apache.flink.client.program.PackagedProgram.callMainMethod(PackagedProgram.java:497)
> ... 6 more
> Caused by: org.apache.flink.client.program.ProgramInvocationException: The
> program execution failed: Could not upload the jar files to the job manager.
> at org.apache.flink.client.program.Client.runBlocking(Client.java:370)
> at org.apache.flink.client.program.Client.runBlocking(Client.java:348)
> at org.apache.flink.client.program.Client.runBlocking(Client.java:315)
> at
> org.apache.flink.client.program.ContextEnvironment.execute(ContextEnvironment.java:70)
> at
> org.apache.flink.api.java.ExecutionEnvironment.execute(ExecutionEnvironment.

Re: Actual byte-streams in multiple-node pipelines

[Fabian Hueske]

Hi Tal,

you said that most processing will be done in external processes. If these
processes are stateful, this might be hard to integrate with Flink's
fault-tolerance mechanism.
In principle, Flink requires two things to achieve exactly-once processing:
1) A data source that can be replayed from a certain point
2) User-functions that can checkpoint and restore their complete state.
(The actual checkpointing is done by Flink, but the user code must expose
its state through Flink's APIs).
In case of a failure, the data sources are set back to a committed point
and the corresponding state of all user functions is restored.

You would need to expose the state of your external functions to Flink and
have some way to reinitialize this state in your external function.

Best,
Fabian


2016-01-20 22:09 GMT+01:00 Stephan Ewen :

> This sounds quite feasible, actually, though it is a pretty unique use
> case.
>
> Like Robert said, you can write map() and flatMap() function on byte[]
> arrays. Make sure that the byte[] that the sources produce are not super
> small and not too large (I would start with 1-4K or so).
>
> You can control how data flows pretty well. It flows 1:1 from produce to
> consumer, if you simply chain these function calls after another. To
> balance bytes across receivers, use rebalance(), broadcast(), or
> partitionCustom().
>
> Streams maintain order of elements, unless steams get split/merged by
> operations like rebalance(), partition() / broadcast / keyBy() or similar.
>
> To union multiple streams with control over how the result stream gets
> pieced together, I would try to connect streams and use a CoMapFunction /
> CoFlatMapFunction to stitch the result stream together form the two input
> streams.
>
> To get exactly-once processing guarantees, activate checkpointing and use
> a source that supports that. If you use a custom source, you may need a few
> lines to integrate it with the checkpoint mechanism, but that is very
> doable.
>
> Hope that helps!
>
> Greetings,
> Stephan
>
>
>
>
> On Wed, Jan 20, 2016 at 2:20 PM, Tal Maoz  wrote:
>
>> Hey Robert,
>>
>> Thanks for responding!
>>
>> The latency I'm talking about would be no more than 1 second from input
>> to output (meaning, bytes should flow immediately through the pipline and
>> get to the other side after going through the processing). You can assume
>> the processors have enough power to work in real-time.
>>
>> The processors would be, for the most part, running external processes
>> (binary executables) and will feed them the incoming data, and then pass
>> along their stdout to the next stage. Simply put, I have several existing
>> 'blackbox' utilities that I need to run on the data in sequence and each of
>> which is a CPU hog...
>>
>> Regarding fault tolerance, no data should be lost and each processor
>> should get the data ONCE and in the correct order (when data is supposed to
>> flow to the same processor). If a node crashes, a new one will take it's
>> place and the data that was sent to the crashed node and was not processed
>> should be sent to the new one, while the output should flow transparently
>> to the next node as if no crashes happened. I know this is a very
>> complicated demand but it is a must in my case.
>>
>> Finally, I'm talking about multiple pipelines running, where each node in
>> a pipeline will be pre-configured before data starts flowing. Each pipeline
>> will read data from a socket or from an MQ if such an MQ exists and is able
>> to handle the load with the required low-latency. Each pipeline's source
>> could be at the range of 45-600MB/s (this can't be split into multiple
>> sources) and eventually, with enough resources and scaling, the system
>> should support hundreds of such pipelines, each with it's own source! Also,
>> at some point, 2 or more sources could be joined with some transformation
>> into a single data stream... Assume the network fabric itself is capable of
>> moving those amounts of data...
>>
>> If i use DataStream where i divide a single segment into very
>> small buffers for low-latency, how can ensure that, on the one hand the
>> data for a single segments flows entirely to the same processor while on
>> the other, different segments can be balanced between several processors?
>>
>>
>> Tal
>>
>> On Wed, Jan 20, 2016 at 3:02 PM, Robert Metzger 
>> wrote:
>>
>>> Hi Tal,
>>>
>>> that sounds like an interesting use case. I think I need a bit more
>>> details about your use case to see how it can be done with Flink.
>>> You said you need low latency, what latency is acceptable for you?
>>>
>>> Also, I was wondering how are you going to feed the input data into
>>> Flink? If the data is coming from multiple sources, maybe everything can be
>>> done completely parallel.
>>> Do you need any fault tolerance guarantees?
>>>
>>> You can use Flink's DataStream abstraction with different data types,
>>> and you could create a DataStream. Flink would internally still send
>>> multi

Re: Unexpected out of bounds error in UnilateralSortMerger

[Theodore Vasiloudis]

This is the stack trace from running with the patched branch:

 The program finished with the following exception:
>
> org.apache.flink.client.program.ProgramInvocationException: The program
> execution failed: Job execution failed.
> at
> org.apache.flink.client.program.Client.runBlocking(Client.java:381)
> at
> org.apache.flink.client.program.Client.runBlocking(Client.java:355)
> at
> org.apache.flink.client.program.Client.runBlocking(Client.java:315)
> at
> org.apache.flink.client.program.ContextEnvironment.execute(ContextEnvironment.java:60)
> at
> org.apache.flink.api.java.ExecutionEnvironment.execute(ExecutionEnvironment.java:803)
> at
> org.apache.flink.api.scala.ExecutionEnvironment.execute(ExecutionEnvironment.scala:591)
> at org.apache.flink.api.scala.DataSet.collect(DataSet.scala:544)
> at fosdem.SVMClassification$.main(SVMClassification.scala:114)
> at fosdem.SVMClassification.main(SVMClassification.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:497)
> at
> org.apache.flink.client.program.PackagedProgram.callMainMethod(PackagedProgram.java:505)
> at
> org.apache.flink.client.program.PackagedProgram.invokeInteractiveModeForExecution(PackagedProgram.java:403)
> at
> org.apache.flink.client.program.Client.runBlocking(Client.java:248)
> at
> org.apache.flink.client.CliFrontend.executeProgramBlocking(CliFrontend.java:796)
> at org.apache.flink.client.CliFrontend.run(CliFrontend.java:323)
> at
> org.apache.flink.client.CliFrontend.parseParameters(CliFrontend.java:1112)
> at org.apache.flink.client.CliFrontend.main(CliFrontend.java:1160)
> Caused by: org.apache.flink.runtime.client.JobExecutionException: Job
> execution failed.
> at
> org.apache.flink.runtime.jobmanager.JobManager$$anonfun$handleMessage$1$$anonfun$applyOrElse$7.apply$mcV$sp(JobManager.scala:659)
> at
> org.apache.flink.runtime.jobmanager.JobManager$$anonfun$handleMessage$1$$anonfun$applyOrElse$7.apply(JobManager.scala:605)
> at
> org.apache.flink.runtime.jobmanager.JobManager$$anonfun$handleMessage$1$$anonfun$applyOrElse$7.apply(JobManager.scala:605)
> at
> scala.concurrent.impl.Future$PromiseCompletingRunnable.liftedTree1$1(Future.scala:24)
> at
> scala.concurrent.impl.Future$PromiseCompletingRunnable.run(Future.scala:24)
> at akka.dispatch.TaskInvocation.run(AbstractDispatcher.scala:41)
> at
> akka.dispatch.ForkJoinExecutorConfigurator$AkkaForkJoinTask.exec(AbstractDispatcher.scala:401)
> at
> scala.concurrent.forkjoin.ForkJoinTask.doExec(ForkJoinTask.java:260)
> at
> scala.concurrent.forkjoin.ForkJoinPool$WorkQueue.pollAndExecAll(ForkJoinPool.java:1253)
> at
> scala.concurrent.forkjoin.ForkJoinPool$WorkQueue.runTask(ForkJoinPool.java:1346)
> at
> scala.concurrent.forkjoin.ForkJoinPool.runWorker(ForkJoinPool.java:1979)
> at
> scala.concurrent.forkjoin.ForkJoinWorkerThread.run(ForkJoinWorkerThread.java:107)
> Caused by: java.lang.RuntimeException: Error obtaining the sorted input:
> Thread 'SortMerger spilling thread' terminated due to an exception:
> java.lang.ArrayIndexOutOfBoundsException
> Serialization trace:
> indices (org.apache.flink.ml.math.SparseVector)
> at
> org.apache.flink.runtime.operators.sort.UnilateralSortMerger.getIterator(UnilateralSortMerger.java:619)
> at
> org.apache.flink.runtime.operators.BatchTask.getInput(BatchTask.java:1085)
> at
> org.apache.flink.runtime.operators.NoOpDriver.run(NoOpDriver.java:78)
> at
> org.apache.flink.runtime.operators.BatchTask.run(BatchTask.java:486)
> at
> org.apache.flink.runtime.operators.BatchTask.invoke(BatchTask.java:351)
> at org.apache.flink.runtime.taskmanager.Task.run(Task.java:567)
> at java.lang.Thread.run(Thread.java:745)
> Caused by: java.io.IOException: Thread 'SortMerger spilling thread'
> terminated due to an exception: java.lang.ArrayIndexOutOfBoundsException
> Serialization trace:
> indices (org.apache.flink.ml.math.SparseVector)
> at
> org.apache.flink.runtime.operators.sort.UnilateralSortMerger$ThreadBase.run(UnilateralSortMerger.java:800)
> Caused by: com.esotericsoftware.kryo.KryoException:
> java.lang.ArrayIndexOutOfBoundsException
> Serialization trace:
> indices (org.apache.flink.ml.math.SparseVector)
> at
> com.esotericsoftware.kryo.serializers.ObjectField.write(ObjectField.java:82)
> at
> com.esotericsoftware.kryo.serializers.FieldSerializer.write(FieldSerializer.java:495)
> at
> com.esotericsoftware.kryo.Kryo.writeClassAndObject(Kryo.java:599)
> at
> org.apache

Re: Cannot start FlinkMiniCluster.WebServer using different port in FlinkMiniCluster

[HungChang]

Thanks for your suggestion. I have some questions to start WebRuntimeMonitor. 

In startWebRuntimeMonitor what should be called for 
- leaderRetrievalService: LeaderRetrievalService,
-  actorSystem: ActorSystem ?

My ref:
(https://ci.apache.org/projects/flink/flink-docs-master/api/java/org/apache/flink/runtime/webmonitor/WebMonitorUtils.html)
Is there any example code using WebRuntimeMonitor?

The file check for the dashboard log passed.

Best,

Sendoh




--
View this message in context: 
http://apache-flink-user-mailing-list-archive.2336050.n4.nabble.com/Cannot-start-FlinkMiniCluster-WebServer-using-different-port-in-FlinkMiniCluster-tp4414p4445.html
Sent from the Apache Flink User Mailing List archive. mailing list archive at 
Nabble.com.

Re: Unexpected out of bounds error in UnilateralSortMerger

[Theodore Vasiloudis]

And this is the one from running with a CSV input, this time I've verified
that I'm using the correct version of Flink, according to Till's
instructions:

 The program finished with the following exception:
>
> org.apache.flink.client.program.ProgramInvocationException: The program
> execution failed: Job execution failed.
> at
> org.apache.flink.client.program.Client.runBlocking(Client.java:381)
> at
> org.apache.flink.client.program.Client.runBlocking(Client.java:355)
> at
> org.apache.flink.client.program.Client.runBlocking(Client.java:315)
> at
> org.apache.flink.client.program.ContextEnvironment.execute(ContextEnvironment.java:60)
> at
> org.apache.flink.api.java.ExecutionEnvironment.execute(ExecutionEnvironment.java:803)
> at
> org.apache.flink.api.scala.ExecutionEnvironment.execute(ExecutionEnvironment.scala:591)
> at org.apache.flink.api.scala.DataSet.collect(DataSet.scala:544)
> at fosdem.SVMClassification$.main(SVMClassification.scala:128)
> at fosdem.SVMClassification.main(SVMClassification.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:497)
> at
> org.apache.flink.client.program.PackagedProgram.callMainMethod(PackagedProgram.java:505)
> at
> org.apache.flink.client.program.PackagedProgram.invokeInteractiveModeForExecution(PackagedProgram.java:403)
> at
> org.apache.flink.client.program.Client.runBlocking(Client.java:248)
> at
> org.apache.flink.client.CliFrontend.executeProgramBlocking(CliFrontend.java:796)
> at org.apache.flink.client.CliFrontend.run(CliFrontend.java:323)
> at
> org.apache.flink.client.CliFrontend.parseParameters(CliFrontend.java:1112)
> at org.apache.flink.client.CliFrontend.main(CliFrontend.java:1160)
> Caused by: org.apache.flink.runtime.client.JobExecutionException: Job
> execution failed.
> at
> org.apache.flink.runtime.jobmanager.JobManager$$anonfun$handleMessage$1$$anonfun$applyOrElse$7.apply$mcV$sp(JobManager.scala:659)
> at
> org.apache.flink.runtime.jobmanager.JobManager$$anonfun$handleMessage$1$$anonfun$applyOrElse$7.apply(JobManager.scala:605)
> at
> org.apache.flink.runtime.jobmanager.JobManager$$anonfun$handleMessage$1$$anonfun$applyOrElse$7.apply(JobManager.scala:605)
> at
> scala.concurrent.impl.Future$PromiseCompletingRunnable.liftedTree1$1(Future.scala:24)
> at
> scala.concurrent.impl.Future$PromiseCompletingRunnable.run(Future.scala:24)
> at akka.dispatch.TaskInvocation.run(AbstractDispatcher.scala:41)
> at
> akka.dispatch.ForkJoinExecutorConfigurator$AkkaForkJoinTask.exec(AbstractDispatcher.scala:401)
> at
> scala.concurrent.forkjoin.ForkJoinTask.doExec(ForkJoinTask.java:260)
> at
> scala.concurrent.forkjoin.ForkJoinPool$WorkQueue.pollAndExecAll(ForkJoinPool.java:1253)
> at
> scala.concurrent.forkjoin.ForkJoinPool$WorkQueue.runTask(ForkJoinPool.java:1346)
> at
> scala.concurrent.forkjoin.ForkJoinPool.runWorker(ForkJoinPool.java:1979)
> at
> scala.concurrent.forkjoin.ForkJoinWorkerThread.run(ForkJoinWorkerThread.java:107)
> Caused by: java.lang.RuntimeException: Error obtaining the sorted input:
> Thread 'SortMerger Reading Thread' terminated due to an exception:
> scala.collection.immutable.Map$EmptyMap$ cannot be cast to
> org.apache.flink.ml.math.Vector
> at
> org.apache.flink.runtime.operators.sort.UnilateralSortMerger.getIterator(UnilateralSortMerger.java:619)
> at
> org.apache.flink.runtime.operators.BatchTask.getInput(BatchTask.java:1085)
> at
> org.apache.flink.runtime.operators.NoOpDriver.run(NoOpDriver.java:78)
> at
> org.apache.flink.runtime.operators.BatchTask.run(BatchTask.java:486)
> at
> org.apache.flink.runtime.operators.BatchTask.invoke(BatchTask.java:351)
> at org.apache.flink.runtime.taskmanager.Task.run(Task.java:567)
> at java.lang.Thread.run(Thread.java:745)
> Caused by: java.io.IOException: Thread 'SortMerger Reading Thread'
> terminated due to an exception: scala.collection.immutable.Map$EmptyMap$
> cannot be cast to org.apache.flink.ml.math.Vector
> at
> org.apache.flink.runtime.operators.sort.UnilateralSortMerger$ThreadBase.run(UnilateralSortMerger.java:799)
> Caused by: java.lang.ClassCastException:
> scala.collection.immutable.Map$EmptyMap$ cannot be cast to
> org.apache.flink.ml.math.Vector
> at
> org.apache.flink.ml.classification.SVM$$anon$25$$anon$11$$anon$1.createInstance(SVM.scala:353)
> at
> org.apache.flink.ml.classification.SVM$$anon$25$$anon$11$$anon$1.createInstance(SVM.scala:353)
> at
> org.apache.flink.api.scala.typeutils.CaseClassSe

Re: Cannot start FlinkMiniCluster.WebServer using different port in FlinkMiniCluster

[Till Rohrmann]

leaderRetrievalService will retrieve the leading JobManager. Take a look at
LeaderRetrievalUtils in order to see how it is created and what options are
supported. actorSystem is the ActorSystem which is used to resolve the
leader’s Akka URL into an ActorRef. You can simply create one or use an
existing one if it is at hand.

I’m wondering why you want to create a WebMonitor yourself?

Cheers,
Till
​

On Thu, Jan 21, 2016 at 12:55 PM, HungChang 
wrote:

> Thanks for your suggestion. I have some questions to start
> WebRuntimeMonitor.
>
> In startWebRuntimeMonitor what should be called for
> - leaderRetrievalService: LeaderRetrievalService,
> -  actorSystem: ActorSystem ?
>
> My ref:
> (
> https://ci.apache.org/projects/flink/flink-docs-master/api/java/org/apache/flink/runtime/webmonitor/WebMonitorUtils.html
> )
> Is there any example code using WebRuntimeMonitor?
>
> The file check for the dashboard log passed.
>
> Best,
>
> Sendoh
>
>
>
>
> --
> View this message in context:
> http://apache-flink-user-mailing-list-archive.2336050.n4.nabble.com/Cannot-start-FlinkMiniCluster-WebServer-using-different-port-in-FlinkMiniCluster-tp4414p4445.html
> Sent from the Apache Flink User Mailing List archive. mailing list archive
> at Nabble.com.
>

Re: Cannot start FlinkMiniCluster.WebServer using different port in FlinkMiniCluster

[HungChang]

Thanks for your reply. 

Yea I'm not sure how to use WebMonitor. For me it's about to write the log
into a file in disk that should go to the job manager originally at
localhost:8081. 

Could you please give an brief example how to use it?

Best,

Sendoh



--
View this message in context: 
http://apache-flink-user-mailing-list-archive.2336050.n4.nabble.com/Cannot-start-FlinkMiniCluster-WebServer-using-different-port-in-FlinkMiniCluster-tp4414p4448.html
Sent from the Apache Flink User Mailing List archive. mailing list archive at 
Nabble.com.

Backpressure in the context of JDBCOutputFormat update

[Maximilian Bode]

Hi everyone,

in a Flink (0.10.1) job with two JDBCOutputFormat sinks, one of them (doing a 
database update) is performing slower than the other one (an insert). The job 
as a whole is also slow as upstream operators are slowed down due to 
backpressure. I am able to speed up the whole job by introducing an a priori 
unnecessary .distinct(), which of course blocks downstream execution of the 
slow sink, which in turn seems to be able to execute faster when given all data 
at once.

Any ideas what is going on here? Is there something I can do without 
introducing unnecessary computation steps?

Cheers,
Max
— 
Maximilian Bode * Junior Consultant * maximilian.b...@tngtech.com * 0176 1000 
75 50
TNG Technology Consulting GmbH, Betastr. 13a, 85774 Unterföhring
Geschäftsführer: Henrik Klagges, Christoph Stock, Dr. Robert Dahlke
Sitz: Unterföhring * Amtsgericht München * HRB 135082

Re: Backpressure in the context of JDBCOutputFormat update

[Robert Metzger]

Hi Max,

is the distinct() operation reducing the size of the DataSet? If so, I
assume you have an idempotent update and the job is faster because fewer
updates are done?
if the distinct() operator is not changing anything, then, the job might be
faster because the INSERT is done while Flink is still executing the
distinct() operation. So the insert is over when the updates are starting.
This would mean that concurrent inserts and updates on the database are
much slower than doing this sequentially.

I'm wondering if there is a way in Flink to explicitly ask for spilling an
intermediate operator to "pause" execution:

Source - > (spill for pausing) ---> (update sink)
\
 --- > (insert)

I don't have a lot of practical experience with RDBMS, but I guess updates
are slower because an index lookup + update is necessary. Maybe optimizing
the database configuration / schema / indexes is more promising. I think
its indeed much nicer to avoid any unnecessary steps in Flink.

Did you do any "microbenchmarks" for the update and insert part? I guess
that would help a lot to understand the impact of certain index structures,
batching sizes, or database drivers.

Regards,
Robert




On Thu, Jan 21, 2016 at 3:35 PM, Maximilian Bode <
maximilian.b...@tngtech.com> wrote:

> Hi everyone,
>
> in a Flink (0.10.1) job with two JDBCOutputFormat sinks, one of them
> (doing a database update) is performing slower than the other one (an
> insert). The job as a whole is also slow as upstream operators are slowed
> down due to backpressure. I am able to speed up the whole job by
> introducing an a priori unnecessary .distinct(), which of course blocks
> downstream execution of the slow sink, which in turn seems to be able to
> execute faster when given all data at once.
>
> Any ideas what is going on here? Is there something I can do without
> introducing unnecessary computation steps?
>
> Cheers,
> Max
> —
> Maximilian Bode * Junior Consultant * maximilian.b...@tngtech.com * 0176
> 1000 75 50
> TNG Technology Consulting GmbH, Betastr. 13a, 85774 Unterföhring
> Geschäftsführer: Henrik Klagges, Christoph Stock, Dr. Robert Dahlke
> Sitz: Unterföhring * Amtsgericht München * HRB 135082
>
>

Re: Backpressure in the context of JDBCOutputFormat update

[Maximilian Bode]

Hi Robert,
sorry, I should have been clearer in my initial mail. The two cases I was 
comparing are:

1) distinct() before Insert (which is necessary as we have a unique key 
constraint in our database), no distinct() before update
2) distinct() before insert AND distinct() before update

The test data used actually only contains unique values for the affected field 
though, so the dataset size is not reduced in case 2.

In case 1 the insert does not start until all the data has arrived at 
distinct() while the update is already going along (slowing down upstream 
operators as well). In case 2 both sinks wait for their respective distinct()'s 
(which is reached much faster now), then start roughly at the same time leading 
to a shorter net job time for job 2 as compared to 1.

A pause operator might be useful, yes.

The update should not be an inherently much more expensive operation, as the 
WHERE clause only contains the table's primary key.

Cheers,
Max
— 
Maximilian Bode * Junior Consultant * maximilian.b...@tngtech.com * 0176 1000 
75 50
TNG Technology Consulting GmbH, Betastr. 13a, 85774 Unterföhring
Geschäftsführer: Henrik Klagges, Christoph Stock, Dr. Robert Dahlke
Sitz: Unterföhring * Amtsgericht München * HRB 135082

> Am 21.01.2016 um 15:57 schrieb Robert Metzger :
> 
> Hi Max,
> 
> is the distinct() operation reducing the size of the DataSet? If so, I assume 
> you have an idempotent update and the job is faster because fewer updates are 
> done?
> if the distinct() operator is not changing anything, then, the job might be 
> faster because the INSERT is done while Flink is still executing the 
> distinct() operation. So the insert is over when the updates are starting. 
> This would mean that concurrent inserts and updates on the database are much 
> slower than doing this sequentially.
> 
> I'm wondering if there is a way in Flink to explicitly ask for spilling an 
> intermediate operator to "pause" execution:
> 
> Source - > (spill for pausing) ---> (update sink)
> \
>  --- > (insert)
> 
> I don't have a lot of practical experience with RDBMS, but I guess updates 
> are slower because an index lookup + update is necessary. Maybe optimizing 
> the database configuration / schema / indexes is more promising. I think its 
> indeed much nicer to avoid any unnecessary steps in Flink.
> 
> Did you do any "microbenchmarks" for the update and insert part? I guess that 
> would help a lot to understand the impact of certain index structures, 
> batching sizes, or database drivers.
> 
> Regards,
> Robert
> 
> 
> 
> 
> On Thu, Jan 21, 2016 at 3:35 PM, Maximilian Bode  > wrote:
> Hi everyone,
> 
> in a Flink (0.10.1) job with two JDBCOutputFormat sinks, one of them (doing a 
> database update) is performing slower than the other one (an insert). The job 
> as a whole is also slow as upstream operators are slowed down due to 
> backpressure. I am able to speed up the whole job by introducing an a priori 
> unnecessary .distinct(), which of course blocks downstream execution of the 
> slow sink, which in turn seems to be able to execute faster when given all 
> data at once.
> 
> Any ideas what is going on here? Is there something I can do without 
> introducing unnecessary computation steps?
> 
> Cheers,
> Max
> — 
> Maximilian Bode * Junior Consultant * maximilian.b...@tngtech.com 
>  * 0176 1000 75 50 
> 
> TNG Technology Consulting GmbH, Betastr. 13a, 85774 Unterföhring
> Geschäftsführer: Henrik Klagges, Christoph Stock, Dr. Robert Dahlke
> Sitz: Unterföhring * Amtsgericht München * HRB 135082
> 
> 

Re: parallelism parameter and output relation

[Serkan Taş]

Hi Robert,

I found the the real reason for the case. Sorry but missed that the example 
project was using 0.8.1.

It is resolved after replacing with 0.10.1.



> 20 Oca 2016 tarihinde 16:40 saatinde, Robert Metzger  
> şunları yazdı:
> 
> Hi Serkan,
> 
> yes, with parallelism=1, you'll get one file, with everything higher, Flink 
> is creating a directory with a file for each parallel instance.
> In your case, Flink can not create (or write to) the file because there is 
> already a directory with the same name. Can you delete the directory and see 
> if writing to the file works afterwards?
> 
> Regards,
> Robert
> 
> 
> 2016-01-20 12:53 GMT+01:00 Serkan Taş  >:
> I am working on this example 
> http://www.itshared.org/2015/03/naive-bayes-on-apache-flink.html 
>  to learn 
> get some more experience on platform.
> 
> Question is ;
> 
> By default, the output of process is double file (named 1 and 2) located in 
> created folder. If i set parallelism to 1, FileNotFound exception is thrown.
> 
> I was expecting to get a single file instead, am i right ?
> 
> 
> Serkan Taş
> Mobil : +90 532 250 07 71 
> Likya Bilgi Teknolojileri
> ve İletişim Hiz. Ltd. Şti.
> www.likyateknoloji.com 
>  
> --
> Bu elektronik posta ve onunla iletilen bütün dosyalar gizlidir. Sadece 
> yukarıda isimleri belirtilen kişiler arasında özel haberleşme amacını 
> taşımaktadır. Size yanlışlıkla ulaşmışsa bu elektonik postanın içeriğini 
> açıklamanız, kopyalamanız, yönlendirmeniz ve kullanmanız kesinlikle yasaktır. 
> Lütfen mesajı geri gönderiniz ve sisteminizden siliniz. Likya Bilgi 
> Teknolojileri ve İletişim Hiz. Ltd. Şti. bu mesajın içeriği ile ilgili olarak 
> hiç bir hukuksal sorumluluğu kabul etmez.
>  
> This electronic mail and any files transmitted with it are intended for the 
> private use of  the persons named above. If you received this message in 
> error, forwarding, copying or use of any of the information is strictly 
> prohibited. Please immediately notify the sender and delete it from your 
> system. Likya Bilgi Teknolojileri ve İletişim Hiz. Ltd. Şti. does not accept 
> legal responsibility for the contents of this message.
> --
> 
> 
> 
> 
> 
> 
> 
> 
> P
> Bu e-postayı yazdırmadan önce, çevreye olan sorumluluğunuzu tekrar düşünün.
> Please consider your environmental responsibility before printing this e-mail.
>  
> 
> 



Serkan Taş
Mobil : +90 532 250 07 71
Likya Bilgi Teknolojileri
ve İletişim Hiz. Ltd. Şti.
www.likyateknoloji.com 
 
--
Bu elektronik posta ve onunla iletilen bütün dosyalar gizlidir. Sadece yukarıda 
isimleri belirtilen kişiler arasında özel haberleşme amacını taşımaktadır. Size 
yanlışlıkla ulaşmışsa bu elektonik postanın içeriğini açıklamanız, 
kopyalamanız, yönlendirmeniz ve kullanmanız kesinlikle yasaktır. Lütfen mesajı 
geri gönderiniz ve sisteminizden siliniz. Likya Bilgi Teknolojileri ve İletişim 
Hiz. Ltd. Şti. bu mesajın içeriği ile ilgili olarak hiç bir hukuksal 
sorumluluğu kabul etmez.
 
This electronic mail and any files transmitted with it are intended for the 
private use of  the persons named above. If you received this message in error, 
forwarding, copying or use of any of the information is strictly prohibited. 
Please immediately notify the sender and delete it from your system. Likya 
Bilgi Teknolojileri ve İletişim Hiz. Ltd. Şti. does not accept legal 
responsibility for the contents of this message.
--








P
Bu e-postayı yazdırmadan önce, çevreye olan sorumluluğunuzu tekrar düşünün.
Please consider your environmental responsibility before printing this e-mail.
 

Re: Cannot start FlinkMiniCluster.WebServer using different port in FlinkMiniCluster

[HungChang]

The following message is obtained after putting BasicConfigurator.configure()
in main();
But I don't understand the reason `flink-runtime-web is not in the
classpath`.

For me the strange part is using the scala version works well whereas my
java version throws exception.

1413 [main] ERROR org.apache.flink.runtime.webmonitor.WebMonitorUtils  -
Could not load web runtime monitor. Probably reason: flink-runtime-web is
not in the classpath
1414 [main] DEBUG org.apache.flink.runtime.webmonitor.WebMonitorUtils  -
Caught exception
java.lang.ClassNotFoundException:
org.apache.flink.runtime.webmonitor.WebRuntimeMonitor
at java.net.URLClassLoader.findClass(URLClassLoader.java:381)
at java.lang.ClassLoader.loadClass(ClassLoader.java:424)
at sun.misc.Launcher$AppClassLoader.loadClass(Launcher.java:331)
at java.lang.ClassLoader.loadClass(ClassLoader.java:357)
at java.lang.Class.forName0(Native Method)
at java.lang.Class.forName(Class.java:264)
at
org.apache.flink.runtime.webmonitor.WebMonitorUtils.startWebRuntimeMonitor(WebMonitorUtils.java:130)
at
org.apache.flink.runtime.minicluster.FlinkMiniCluster.startWebServer(FlinkMiniCluster.scala:292)
at
org.apache.flink.runtime.minicluster.FlinkMiniCluster.start(FlinkMiniCluster.scala:268)
at
org.apache.flink.runtime.minicluster.FlinkMiniCluster.start(FlinkMiniCluster.scala:226)
at
org.apache.flink.streaming.api.environment.LocalStreamEnvironment.execute(LocalStreamEnvironment.java:101)

Best,

Sendoh




--
View this message in context: 
http://apache-flink-user-mailing-list-archive.2336050.n4.nabble.com/Cannot-start-FlinkMiniCluster-WebServer-using-different-port-in-FlinkMiniCluster-tp4414p4453.html
Sent from the Apache Flink User Mailing List archive. mailing list archive at 
Nabble.com.

Re: Cannot start FlinkMiniCluster.WebServer using different port in FlinkMiniCluster

[Till Rohrmann]

Could you add flink-runtime-web to your dependencies of your project? It
seems as if it is missing in your project.

Cheers,
Till
​

On Thu, Jan 21, 2016 at 4:45 PM, HungChang 
wrote:

> The following message is obtained after putting
> BasicConfigurator.configure()
> in main();
> But I don't understand the reason `flink-runtime-web is not in the
> classpath`.
>
> For me the strange part is using the scala version works well whereas my
> java version throws exception.
>
> 1413 [main] ERROR org.apache.flink.runtime.webmonitor.WebMonitorUtils  -
> Could not load web runtime monitor. Probably reason: flink-runtime-web is
> not in the classpath
> 1414 [main] DEBUG org.apache.flink.runtime.webmonitor.WebMonitorUtils  -
> Caught exception
> java.lang.ClassNotFoundException:
> org.apache.flink.runtime.webmonitor.WebRuntimeMonitor
> at java.net.URLClassLoader.findClass(URLClassLoader.java:381)
> at java.lang.ClassLoader.loadClass(ClassLoader.java:424)
> at sun.misc.Launcher$AppClassLoader.loadClass(Launcher.java:331)
> at java.lang.ClassLoader.loadClass(ClassLoader.java:357)
> at java.lang.Class.forName0(Native Method)
> at java.lang.Class.forName(Class.java:264)
> at
>
> org.apache.flink.runtime.webmonitor.WebMonitorUtils.startWebRuntimeMonitor(WebMonitorUtils.java:130)
> at
>
> org.apache.flink.runtime.minicluster.FlinkMiniCluster.startWebServer(FlinkMiniCluster.scala:292)
> at
>
> org.apache.flink.runtime.minicluster.FlinkMiniCluster.start(FlinkMiniCluster.scala:268)
> at
>
> org.apache.flink.runtime.minicluster.FlinkMiniCluster.start(FlinkMiniCluster.scala:226)
> at
>
> org.apache.flink.streaming.api.environment.LocalStreamEnvironment.execute(LocalStreamEnvironment.java:101)
>
> Best,
>
> Sendoh
>
>
>
>
> --
> View this message in context:
> http://apache-flink-user-mailing-list-archive.2336050.n4.nabble.com/Cannot-start-FlinkMiniCluster-WebServer-using-different-port-in-FlinkMiniCluster-tp4414p4453.html
> Sent from the Apache Flink User Mailing List archive. mailing list archive
> at Nabble.com.
>

Re: Cannot start FlinkMiniCluster.WebServer using different port in FlinkMiniCluster

[HungChang]

After adding the dependency it totally works! Thank you a lot!



--
View this message in context: 
http://apache-flink-user-mailing-list-archive.2336050.n4.nabble.com/Cannot-start-FlinkMiniCluster-WebServer-using-different-port-in-FlinkMiniCluster-tp4414p4455.html
Sent from the Apache Flink User Mailing List archive. mailing list archive at 
Nabble.com.

Re: Cannot start FlinkMiniCluster.WebServer using different port in FlinkMiniCluster

[Till Rohrmann]

Great to hear :-)

On Thu, Jan 21, 2016 at 4:55 PM, HungChang 
wrote:

> After adding the dependency it totally works! Thank you a lot!
>
>
>
> --
> View this message in context:
> http://apache-flink-user-mailing-list-archive.2336050.n4.nabble.com/Cannot-start-FlinkMiniCluster-WebServer-using-different-port-in-FlinkMiniCluster-tp4414p4455.html
> Sent from the Apache Flink User Mailing List archive. mailing list archive
> at Nabble.com.
>

Re: Results of testing Flink quickstart against 0.10-SNAPSHOT and 1.0-SNAPSHOT (re. Dependency on non-existent org.scalamacros:quasiquotes_2.11:)

[Prez Cannady]

Apologies for the late reply; been on the road.

I’ve been blowing away my Maven and Gradle repos for each test to make sure I’m 
pulling down the latest snapshots.  I’ve also looked into the source jars 
directly from the snapshot repo:

1.0-SNAPSHOT: 
https://repository.apache.org/content/repositories/snapshots/org/apache/flink/flink-runtime_2.11/1.0-SNAPSHOT/
 

0.10-SNAPSHOT: 
https://repository.apache.org/content/repositories/snapshots/org/apache/flink/flink-runtime_2.11/0.10-SNAPSHOT/
 


I note that the jars date to January 14, which may be right on the cusp of when 
this commit was taken (7 days ago).  And while the change you and I first 
discussed made it 
(https://github.com/apache/flink/commit/0ae46b596949808f56c40bd7a68f478bc10206ab
 
),
 the second did not 
(https://github.com/apache/flink/commit/cfcb5d7ba5c22295c0ae628f8d9a2447e2286985
 
).

Perhaps there’s a fresher repo I should be using?

Prez Cannady  
p: 617 500 3378  
e: revp...@opencorrelate.org   
GH: https://github.com/opencorrelate   
LI: https://www.linkedin.com/in/revprez   









> On Jan 20, 2016, at 3:48 PM, Stephan Ewen  wrote:
> 
> Hi Prez!
> 
> I merged the pull request into master a while back. Have a look here 
> (https://github.com/apache/flink/commits/master 
>  commits of January 15th).
> 
> Is it possible that you are using a cached older version?
> 
> Greetings,
> Stephan
> 
> 
> 
> 
> On Wed, Jan 20, 2016 at 4:00 PM, Prez Cannady  > wrote:
> Morning, Robert.
> 
> You’re right; the 1.0-SNAPSHOT with fetched binaries issue is resolved now.  
> Unfortunately, it now emits the same error as 0.10-SNAPSHOT with fetched 
> binaries.  There is a fix for that:
> 
> https://github.com/apache/flink/pull/1511 
> 
> 
> It’s definitely in the release-0.10 and master branches
> 
> https://github.com/apache/flink/blob/master/flink-runtime/src/main/java/org/apache/flink/runtime/accumulators/AccumulatorSnapshot.java
>  
> 
> https://github.com/apache/flink/blob/release-0.10/flink-runtime/src/main/java/org/apache/flink/runtime/accumulators/AccumulatorSnapshot.java
>  
> 
> 
> I grabbed the sources jar for the latest 1.0-SNAPSHOT and 0.10-SNAPSHOT:
> 
> From 
> https://repository.apache.org/content/repositories/snapshots/org/apache/flink/flink-runtime_2.11/0.10-SNAPSHOT/flink-runtime_2.11-0.10-20160114.200924-161-sources.jar
>  
> 
> 
> /**
>  * Gets the Flink (internal) accumulators values.
>  * @return the serialized map
>  */
> public Map> 
> deserializeFlinkAccumulators() throws IOException, ClassNotFoundException {
> return 
> flinkAccumulators.deserializeValue(ClassLoader.getSystemClassLoader());
> }
> 
> ClassLoader.getSystemClassLoader() should be getClass().getClassLoader().
> 
> Not sure why it’s not taking in the build, but there’s the problem.
> 
> Prez Cannady  
> p: 617 500 3378  
> e: revp...@opencorrelate.org   
> GH: https://github.com/opencorrelate   
> LI: https://www.linkedin.com/in/revprez  
>  
> 
> 
> 
> 
> 
> 
> 
> 
> 
>> On Jan 20, 2016, at 8:17 AM, Robert Metzger > > wrote:
>> 
>> Hi Prez,
>> 
>> thanks a lot for the thorough research you did on this issue. The issue with 
>> "1.0-SNAPSHOT with fetched binary dependencies" should be resolved by a fix 
>> I've pushed to master yesterday:
>> 
>> a) The "change-scala-version" script wasn't adopted to the renamed examples 
>> directory, that's why it failed renaming the artifacts for _2.11. That's why 
>> the maven dependencies ended up being mixed between Scala 2.11 and 2.10.
>> https://github.com/apache/flink/commit/8f0c47df092ccdc6028dbd88aed197edcd8945ee#diff-c1ca4095c51fcc58f380c59cfabffc8a
>>  
>> 
>> 
>> b) The deployment of the s

Re: Actual byte-streams in multiple-node pipelines

[Tal Maoz]

Thanks Stephan and Fabian!

You make very valuable points! This really helps steer me in the right
direction!
It would take some more careful planning and implementing the components
you suggested but hopefully it will work in the end...

Thanks,

Tal

On Thu, Jan 21, 2016 at 11:20 AM, Fabian Hueske  wrote:

> Hi Tal,
>
> you said that most processing will be done in external processes. If these
> processes are stateful, this might be hard to integrate with Flink's
> fault-tolerance mechanism.
> In principle, Flink requires two things to achieve exactly-once processing:
> 1) A data source that can be replayed from a certain point
> 2) User-functions that can checkpoint and restore their complete state.
> (The actual checkpointing is done by Flink, but the user code must expose
> its state through Flink's APIs).
> In case of a failure, the data sources are set back to a committed point
> and the corresponding state of all user functions is restored.
>
> You would need to expose the state of your external functions to Flink and
> have some way to reinitialize this state in your external function.
>
> Best,
> Fabian
>
>
> 2016-01-20 22:09 GMT+01:00 Stephan Ewen :
>
>> This sounds quite feasible, actually, though it is a pretty unique use
>> case.
>>
>> Like Robert said, you can write map() and flatMap() function on byte[]
>> arrays. Make sure that the byte[] that the sources produce are not super
>> small and not too large (I would start with 1-4K or so).
>>
>> You can control how data flows pretty well. It flows 1:1 from produce to
>> consumer, if you simply chain these function calls after another. To
>> balance bytes across receivers, use rebalance(), broadcast(), or
>> partitionCustom().
>>
>> Streams maintain order of elements, unless steams get split/merged by
>> operations like rebalance(), partition() / broadcast / keyBy() or similar.
>>
>> To union multiple streams with control over how the result stream gets
>> pieced together, I would try to connect streams and use a CoMapFunction /
>> CoFlatMapFunction to stitch the result stream together form the two input
>> streams.
>>
>> To get exactly-once processing guarantees, activate checkpointing and use
>> a source that supports that. If you use a custom source, you may need a few
>> lines to integrate it with the checkpoint mechanism, but that is very
>> doable.
>>
>> Hope that helps!
>>
>> Greetings,
>> Stephan
>>
>>
>>
>>
>> On Wed, Jan 20, 2016 at 2:20 PM, Tal Maoz  wrote:
>>
>>> Hey Robert,
>>>
>>> Thanks for responding!
>>>
>>> The latency I'm talking about would be no more than 1 second from input
>>> to output (meaning, bytes should flow immediately through the pipline and
>>> get to the other side after going through the processing). You can assume
>>> the processors have enough power to work in real-time.
>>>
>>> The processors would be, for the most part, running external processes
>>> (binary executables) and will feed them the incoming data, and then pass
>>> along their stdout to the next stage. Simply put, I have several existing
>>> 'blackbox' utilities that I need to run on the data in sequence and each of
>>> which is a CPU hog...
>>>
>>> Regarding fault tolerance, no data should be lost and each processor
>>> should get the data ONCE and in the correct order (when data is supposed to
>>> flow to the same processor). If a node crashes, a new one will take it's
>>> place and the data that was sent to the crashed node and was not processed
>>> should be sent to the new one, while the output should flow transparently
>>> to the next node as if no crashes happened. I know this is a very
>>> complicated demand but it is a must in my case.
>>>
>>> Finally, I'm talking about multiple pipelines running, where each node
>>> in a pipeline will be pre-configured before data starts flowing. Each
>>> pipeline will read data from a socket or from an MQ if such an MQ exists
>>> and is able to handle the load with the required low-latency. Each
>>> pipeline's source could be at the range of 45-600MB/s (this can't be split
>>> into multiple sources) and eventually, with enough resources and scaling,
>>> the system should support hundreds of such pipelines, each with it's own
>>> source! Also, at some point, 2 or more sources could be joined with some
>>> transformation into a single data stream... Assume the network fabric
>>> itself is capable of moving those amounts of data...
>>>
>>> If i use DataStream where i divide a single segment into very
>>> small buffers for low-latency, how can ensure that, on the one hand the
>>> data for a single segments flows entirely to the same processor while on
>>> the other, different segments can be balanced between several processors?
>>>
>>>
>>> Tal
>>>
>>> On Wed, Jan 20, 2016 at 3:02 PM, Robert Metzger 
>>> wrote:
>>>
 Hi Tal,

 that sounds like an interesting use case. I think I need a bit more
 details about your use case to see how it can be done with Flink.
 You said you need low latenc

Re: DeserializationSchema isEndOfStream usage?

[David Kim]

Hi Robert!

Thanks for reaching out. I ran into an issue and wasn't sure if this was
due to a misconfiguration on my end of if this is a real bug. I have one
DataStream and I'm sinking to two different kafka sinks. When the job
starts, I run into this error:

org.apache.flink.runtime.client.JobExecutionException: Job execution failed.
at
org.apache.flink.runtime.jobmanager.JobManager$$anonfun$handleMessage$1$$anonfun$applyOrElse$7.apply$mcV$sp(JobManager.scala:659)
at
org.apache.flink.runtime.jobmanager.JobManager$$anonfun$handleMessage$1$$anonfun$applyOrElse$7.apply(JobManager.scala:605)
at
org.apache.flink.runtime.jobmanager.JobManager$$anonfun$handleMessage$1$$anonfun$applyOrElse$7.apply(JobManager.scala:605)
at
scala.concurrent.impl.Future$PromiseCompletingRunnable.liftedTree1$1(Future.scala:24)
at
scala.concurrent.impl.Future$PromiseCompletingRunnable.run(Future.scala:24)
at akka.dispatch.TaskInvocation.run(AbstractDispatcher.scala:41)
at
akka.dispatch.ForkJoinExecutorConfigurator$AkkaForkJoinTask.exec(AbstractDispatcher.scala:401)
at scala.concurrent.forkjoin.ForkJoinTask.doExec(ForkJoinTask.java:260)
at
scala.concurrent.forkjoin.ForkJoinPool$WorkQueue.runTask(ForkJoinPool.java:1339)
at scala.concurrent.forkjoin.ForkJoinPool.runWorker(ForkJoinPool.java:1979)
at
scala.concurrent.forkjoin.ForkJoinWorkerThread.run(ForkJoinWorkerThread.java:107)
Caused by: java.lang.UnsupportedOperationException: The accumulator
'producer-record-retry-rate' already exists and cannot be added.
at
org.apache.flink.api.common.functions.util.AbstractRuntimeUDFContext.addAccumulator(AbstractRuntimeUDFContext.java:121)
at
org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducerBase.open(FlinkKafkaProducerBase.java:204)
at
org.apache.flink.api.common.functions.util.FunctionUtils.openFunction(FunctionUtils.java:36)
at
org.apache.flink.streaming.api.operators.AbstractUdfStreamOperator.open(AbstractUdfStreamOperator.java:89)
at
org.apache.flink.streaming.runtime.tasks.StreamTask.openAllOperators(StreamTask.java:305)
at
org.apache.flink.streaming.runtime.tasks.StreamTask.invoke(StreamTask.java:227)
at org.apache.flink.runtime.taskmanager.Task.run(Task.java:567)
at java.lang.Thread.run(Thread.java:745)


The particular accumulator the exception is complains about changes,
meaning it's not always 'producer-record-retry-rate' -- most likely due to
the non-deterministic ordering of the collection. Any guidance appreciated!

I'm using 1.0-SNAPSHOT and my two sinks are FlinkKafkaProducer08.

The flink code looks something like this:


val stream: DataStream[Foo] = ...

val kafkaA = new FlinkKafkaProducer08[Foo]...

val kafkaB = new FlinkKafkaProducer08[Foo]...


stream
  .addSink(kafkaA)

stream.
  .addSink(kafkaB)


Thanks,
David

On Wed, Jan 20, 2016 at 1:34 PM, Robert Metzger  wrote:

> I've now merged the pull request. DeserializationSchema.isEndOfStream()
> should now be evaluated correctly by the Kafka 0.9 and 0.8 connectors.
>
> Please let me know if the updated code has any issues. I'll fix the issues
> asap.
>
> On Wed, Jan 13, 2016 at 5:06 PM, David Kim <
> david@braintreepayments.com> wrote:
>
>> Thanks Robert! I'll be keeping tabs on the PR.
>>
>> Cheers,
>> David
>>
>> On Mon, Jan 11, 2016 at 4:04 PM, Robert Metzger 
>> wrote:
>>
>>> Hi David,
>>>
>>> In theory isEndOfStream() is absolutely the right way to go for stopping
>>> data sources in Flink.
>>> That its not working as expected is a bug. I have a pending pull request
>>> for adding a Kafka 0.9 connector, which fixes this issue as well (for all
>>> supported Kafka versions).
>>>
>>> Sorry for the inconvenience. If you want, you can check out the branch
>>> of the PR and build Flink yourself to get the fix.
>>> I hope that I can merge the connector to master this week, then, the fix
>>> will be available in 1.0-SNAPSHOT as well.
>>>
>>> Regards,
>>> Robert
>>>
>>>
>>>
>>> Sent from my iPhone
>>>
>>> On 11.01.2016, at 21:39, David Kim 
>>> wrote:
>>>
>>> Hello all,
>>>
>>> I saw that DeserializationSchema has an API "isEndOfStream()".
>>>
>>>
>>> https://github.com/apache/flink/blob/master/flink-streaming-java/src/main/java/org/apache/flink/streaming/util/serialization/DeserializationSchema.java
>>>
>>> Can *isEndOfStream* be utilized to somehow terminate a streaming flink
>>> job?
>>>
>>> I was under the impression that if we return "true" we can control when
>>> a stream can close. The use case I had in mind was controlling when
>>> unit/integration tests would terminate a flink job. We can rely on the fact
>>> that a test/spec would know how many items it expects to consume and then
>>> switch *isEndOfStream* to return true.
>>>
>>> Am I misunderstanding the intention for *isEndOfStream*?
>>>
>>> I also set a breakpoint on *isEndOfStream* and saw that it never was
>>> hit when using "FlinkKafkaConsumer082" to pass in a DeserializationSchema
>>> implementation.
>>>
>>> Currently testing on 1.0-SNAPSHOT.
>>>
>>> Cheers!
>>> David
>>>
>>>
>>
>>
>> --
>> Note: