2015-12-02 20 views
5

क्या मैं फ्लिंक में हमारे कार्यक्रम में कार्य के विभिन्न हिस्सों के लिए समानांतरता की विभिन्न डिग्री निर्धारित कर सकता हूं? उदाहरण के लिए, फ्लिंक निम्नलिखित नमूना कोड की व्याख्या कैसे करता है? दो कस्टम प्रैक्टिशनर्स MyPartitioner1, MyPartitioner2, इनपुट डेटा को दो 4 और 2 विभाजन विभाजित करते हैं।अपाचे फ्लिंक में समांतरता की डिग्री

partitionedData1 = inputData1 
    .partitionCustom(new MyPartitioner1(), 1); 
env.setParallelism(4); 
DataSet<Tuple2<Integer, Integer>> output1 = partitionedData1 
    .mapPartition(new calculateFun()); 

partitionedData2 = inputData2 
    .partitionCustom(new MyPartitioner2(), 2); 
env.setParallelism(2); 
DataSet<Tuple2<Integer, Integer>> output2 = partitionedData2 
    .mapPartition(new calculateFun()); 

मैं इस कोड के लिए निम्न त्रुटि मिलती है:

Exception in thread "main" org.apache.flink.runtime.client.JobExecutionException: Job execution failed. 
    at org.apache.flink.runtime.jobmanager.JobManager$$anonfun$receiveWithLogMessages$1.applyOrElse(JobManager.scala:314) 
    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.ActorLogMessages$$anon$1.apply(ActorLogMessages.scala:36) 
    at org.apache.flink.runtime.ActorLogMessages$$anon$1.apply(ActorLogMessages.scala:29) 
    at scala.PartialFunction$class.applyOrElse(PartialFunction.scala:118) 
    at org.apache.flink.runtime.ActorLogMessages$$anon$1.applyOrElse(ActorLogMessages.scala:29) 
    at akka.actor.Actor$class.aroundReceive(Actor.scala:465) 
    at org.apache.flink.runtime.jobmanager.JobManager.aroundReceive(JobManager.scala:92) 
    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.lang.ArrayIndexOutOfBoundsException: 2 
    at org.apache.flink.runtime.io.network.api.writer.RecordWriter.emit(RecordWriter.java:80) 
    at org.apache.flink.runtime.operators.shipping.OutputCollector.collect(OutputCollector.java:65) 
    at org.apache.flink.runtime.operators.NoOpDriver.run(NoOpDriver.java:92) 
    at org.apache.flink.runtime.operators.RegularPactTask.run(RegularPactTask.java:496) 
    at org.apache.flink.runtime.operators.RegularPactTask.invoke(RegularPactTask.java:362) 
    at org.apache.flink.runtime.taskmanager.Task.run(Task.java:559) 
    at java.lang.Thread.run(Unknown Source) 

उत्तर

5

ExecutionEnvironment.setParallelism() पूरे कार्यक्रम, जैसे कि, कार्यक्रम के सभी ऑपरेटरों के लिए समानांतरवाद सेट।

आप ऑपरेटर पर setParallelism() विधि को कॉल करके प्रत्येक व्यक्तिगत ऑपरेटर के लिए समांतरता निर्दिष्ट कर सकते हैं।

ArrayIndexOutOfBoundsException फेंक दिया गया है क्योंकि आपका कस्टम विभाजनकर्ता समानांतरता की अप्रत्याशित डिग्री के कारण शायद एक अवैध विभाजन संख्या देता है। कस्टम विभाजनकर्ता को partition(K key, int numPartitions) विधि में पैरामीटर के रूप में रिसीवर की वास्तविक समांतरता प्राप्त होती है।

संबंधित मुद्दे