Hi Till,

thanks for your reply!
I tested it with the Wordcount example.
Everything works fine if I run the command:
./flink run -p 3 /home/flink/examples/WordCount.jar
Then the program gets executed by my 3 workers. 
If I want to save the output to a file:
./flink run -p 3 /home/flink/examples/WordCount.jar 
hdfs://grips2:9000/users/Flink_1000.csv hdfs://grips2:9000/users/Wordcount_1000 
<hdfs://grips2:9000/users/Wordcount_1000>

I get the following error message:
What am I doing wrong? Is something wrong with my cluster writing permissions?

org.apache.flink.client.program.ProgramInvocationException: The program 
execution failed: Cannot initialize task 'DataSink (CsvOutputFormat (path: 
hdfs://grips2:9000/users/Wordcount_1000s, delimiter:  ))': Output directory 
could not be created.
        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.examples.java.wordcount.WordCount.main(WordCount.java:78)
        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:483)
        at 
org.apache.flink.client.program.PackagedProgram.callMainMethod(PackagedProgram.java:497)
        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:676)
        at org.apache.flink.client.CliFrontend.run(CliFrontend.java:326)
        at 
org.apache.flink.client.CliFrontend.parseParameters(CliFrontend.java:978)
        at org.apache.flink.client.CliFrontend.main(CliFrontend.java:1028)
Caused by: org.apache.flink.runtime.client.JobExecutionException: Cannot 
initialize task 'DataSink (CsvOutputFormat (path: 
hdfs://grips2:9000/users/Wordcount_1000s, delimiter:  ))': Output directory 
could not be created.
        at 
org.apache.flink.runtime.jobmanager.JobManager$$anonfun$org$apache$flink$runtime$jobmanager$JobManager$$submitJob$6.apply(JobManager.scala:867)
        at 
org.apache.flink.runtime.jobmanager.JobManager$$anonfun$org$apache$flink$runtime$jobmanager$JobManager$$submitJob$6.apply(JobManager.scala:851)
        at scala.collection.Iterator$class.foreach(Iterator.scala:727)
        at scala.collection.AbstractIterator.foreach(Iterator.scala:1157)
        at scala.collection.IterableLike$class.foreach(IterableLike.scala:72)
        at scala.collection.AbstractIterable.foreach(Iterable.scala:54)
        at 
org.apache.flink.runtime.jobmanager.JobManager.org$apache$flink$runtime$jobmanager$JobManager$$submitJob(JobManager.scala:851)
        at 
org.apache.flink.runtime.jobmanager.JobManager$$anonfun$handleMessage$1.applyOrElse(JobManager.scala:341)
        at 
scala.runtime.AbstractPartialFunction$mcVL$sp.apply$mcVL$sp(AbstractPartialFunction.scala:33)
        at 
scala.runtime.AbstractPartialFunction$mcVL$sp.apply(AbstractPartialFunction.scala:33)
        at 
scala.runtime.AbstractPartialFunction$mcVL$sp.apply(AbstractPartialFunction.scala:25)
        at 
org.apache.flink.runtime.LeaderSessionMessageFilter$$anonfun$receive$1.applyOrElse(LeaderSessionMessageFilter.scala:36)
        at 
scala.runtime.AbstractPartialFunction$mcVL$sp.apply$mcVL$sp(AbstractPartialFunction.scala:33)
        at 
scala.runtime.AbstractPartialFunction$mcVL$sp.apply(AbstractPartialFunction.scala:33)
        at 
scala.runtime.AbstractPartialFunction$mcVL$sp.apply(AbstractPartialFunction.scala:25)
        at 
org.apache.flink.runtime.LogMessages$$anon$1.apply(LogMessages.scala:33)
        at 
org.apache.flink.runtime.LogMessages$$anon$1.apply(LogMessages.scala:28)
        at scala.PartialFunction$class.applyOrElse(PartialFunction.scala:118)
        at 
org.apache.flink.runtime.LogMessages$$anon$1.applyOrElse(LogMessages.scala:28)
        at akka.actor.Actor$class.aroundReceive(Actor.scala:465)
        at 
org.apache.flink.runtime.jobmanager.JobManager.aroundReceive(JobManager.scala:100)
        at akka.actor.ActorCell.receiveMessage(ActorCell.scala:516)
        at akka.actor.ActorCell.invoke(ActorCell.scala:487)
        at akka.dispatch.Mailbox.processMailbox(Mailbox.scala:254)
        at akka.dispatch.Mailbox.run(Mailbox.scala:221)
        at akka.dispatch.Mailbox.exec(Mailbox.scala:231)
        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.io.IOException: Output directory could not be created.
        at 
org.apache.flink.api.common.io.FileOutputFormat.initializeGlobal(FileOutputFormat.java:295)
        at 
org.apache.flink.runtime.jobgraph.OutputFormatVertex.initializeOnMaster(OutputFormatVertex.java:84)
        at 
org.apache.flink.runtime.jobmanager.JobManager$$anonfun$org$apache$flink$runtime$jobmanager$JobManager$$submitJob$6.apply(JobManager.scala:863)
        ... 29 more

The exception above occurred while trying to run your command.


> Am 28.01.2016 um 10:44 schrieb Till Rohrmann <till.rohrm...@gmail.com>:
> 
> Hi Lydia,
> 
> what do you mean with master? Usually when you submit a program to the 
> cluster and don’t specify the parallelism in your program, then it will be 
> executed with the parallelism.default value as parallelism. You can specify 
> the value in your cluster configuration flink-config.yaml file. Alternatively 
> you can always specify the parallelism via the CLI client with the -p option.
> 
> Cheers,
> Till
> 
> 
> On Thu, Jan 28, 2016 at 9:53 AM, Lydia Ickler <ickle...@googlemail.com 
> <mailto:ickle...@googlemail.com>> wrote:
> Hi all,
> 
> I am doing some operations on a DataSet<Tuple3<Integer,Integer,Double>> … 
> (see code below)
> When I run my program on a cluster with 3 machines I can see within the web 
> client that only my master is executing the program. 
> Do I have to specify somewhere that all machines have to participate? Usually 
> the cluster executes in parallel.
> 
> Any suggestions?
> 
> Best regards, 
> Lydia
> DataSet<Tuple3<Integer, Integer, Double>> matrixA = readMatrix(env, input);
> DataSet<Tuple3<Integer, Integer, Double>> initial = matrixA.groupBy(0).sum(2);
> 
> //normalize by maximum value
> initial = initial.cross(initial.max(2)).map(new normalizeByMax());
> matrixA.join(initial).where(1).equalTo(0)
>       .map(new ProjectJoinResultMapper()).groupBy(0, 1).sum(2);
> 
> 

Reply via email to