Thanks allot.
I switched to flink0.6.1
Now i get a slightly different exception: "The channel is erroneous."
...
...rojectFlatJoinFunction) (159/162) switched to CANCELED
10/08/2014 20:27:58:
Join(org.apache.flink.api.java.operators.JoinOperator$ProjectFlatJoinFunction)
(160/162) switched to CANCELED
10/08/2014 20:27:58:
Join(org.apache.flink.api.java.operators.JoinOperator$ProjectFlatJoinFunction)
(161/162) switched to CANCELED
10/08/2014 20:27:58: Job execution switched to status FAILED
Error: The program execution failed: java.lang.Exception: The data
preparation for task
'Reduce(org.apache.flink.allIn.StratosphereMultiFlink6Job$ReduceSim2)' ,
caused an error: Error obtaining the sorted input: Thread 'SortMerger
spilling thread' terminated due to an exception: The channel is erroneous.
at
org.apache.flink.runtime.operators.RegularPactTask.run(RegularPactTask.java:485)
at
org.apache.flink.runtime.operators.RegularPactTask.invoke(RegularPactTask.java:375)
at
org.apache.flink.runtime.execution.RuntimeEnvironment.run(RuntimeEnvironment.java:265)
at java.lang.Thread.run(Thread.java:744)
Caused by: java.lang.RuntimeException: Error obtaining the sorted input:
Thread 'SortMerger spilling thread' terminated due to an exception: The
channel is erroneous.
at
org.apache.flink.runtime.operators.sort.UnilateralSortMerger.getIterator(UnilateralSortMerger.java:616)
at
org.apache.flink.runtime.operators.RegularPactTask.getInput(RegularPactTask.java:1131)
at
org.apache.flink.runtime.operators.GroupReduceDriver.prepare(GroupReduceDriver.java:91)
at
org.apache.flink.runtime.operators.RegularPactTask.run(RegularPactTask.java:480)
... 3 more
Caused by: java.io.IOException: Thread 'SortMerger spilling thread'
terminated due to an exception: The channel is erroneous.
at
org.apache.flink.runtime.operators.sort.UnilateralSortMerger$ThreadBase.run(UnilateralSortMerger.java:801)
Caused by: java.io.IOException: The channel is erroneous.
at
org.apache.flink.runtime.io.disk.iomanager.ChannelAccess.checkErroneous(ChannelAccess.java:132)
at
org.apache.flink.runtime.io.disk.iomanager.BlockChannelWriter.writeBlock(BlockChannelWriter.java:73)
at
org.apache.flink.runtime.io.disk.iomanager.ChannelWriterOutputView.writeSegment(ChannelWriterOutputView.java:220)
at
org.apache.flink.runtime.io.disk.iomanager.ChannelWriterOutputView.nextSegment(ChannelWriterOutputView.java:206)
at
org.apache.flink.runtime.memorymanager.AbstractPagedOutputView.advance(AbstractPagedOutputView.java:140)
at
org.apache.flink.runtime.memorymanager.AbstractPagedOutputView.writeInt(AbstractPagedOutputView.java:267)
at
org.apache.flink.api.common.typeutils.base.IntSerializer.copy(IntSerializer.java:74)
at
org.apache.flink.api.java.typeutils.runtime.TupleSerializer.copy(TupleSerializer.java:124)
at
org.apache.flink.runtime.operators.sort.NormalizedKeySorter.writeToOutput(NormalizedKeySorter.java:449)
at
org.apache.flink.runtime.operators.sort.UnilateralSortMerger$SpillingThread.go(UnilateralSortMerger.java:1316)
at
org.apache.flink.runtime.operators.sort.UnilateralSortMerger$ThreadBase.run(UnilateralSortMerger.java:798)
Caused by: java.io.IOException: No space left on device
at sun.nio.ch.FileDispatcherImpl.write0(Native Method)
at sun.nio.ch.FileDispatcherImpl.write(FileDispatcherImpl.java:60)
at sun.nio.ch.IOUtil.writeFromNativeBuffer(IOUtil.java:93)
at sun.nio.ch.IOUtil.write(IOUtil.java:65)
at sun.nio.ch.FileChannelImpl.write(FileChannelImpl.java:205)
at
org.apache.flink.runtime.io.disk.iomanager.SegmentWriteRequest.write(BlockChannelAccess.java:259)
at
org.apache.flink.runtime.io.disk.iomanager.IOManager$WriterThread.run(IOManager.java:644)
org.apache.flink.client.program.ProgramInvocationException: The program
execution failed: java.lang.Exception: The data preparation for task
'Reduce(org.apache.flink.allIn.StratosphereMultiFlink6Job$ReduceSim2)' ,
caused an error: Error obtaining the sorted input: Thread 'SortMerger
spilling thread' terminated due to an exception: The channel is erroneous.
at
org.apache.flink.runtime.operators.RegularPactTask.run(RegularPactTask.java:485)
at
org.apache.flink.runtime.operators.RegularPactTask.invoke(RegularPactTask.java:375)
at
org.apache.flink.runtime.execution.RuntimeEnvironment.run(RuntimeEnvironment.java:265)
at java.lang.Thread.run(Thread.java:744)
Caused by: java.lang.RuntimeException: Error obtaining the sorted input:
Thread 'SortMerger spilling thread' terminated due to an exception: The
channel is erroneous.
at
org.apache.flink.runtime.operators.sort.UnilateralSortMerger.getIterator(UnilateralSortMerger.java:616)
at
org.apache.flink.runtime.operators.RegularPactTask.getInput(RegularPactTask.java:1131)
at
org.apache.flink.runtime.operators.GroupReduceDriver.prepare(GroupReduceDriver.java:91)
at
org.apache.flink.runtime.operators.RegularPactTask.run(RegularPactTask.java:480)
... 3 more
Caused by: java.io.IOException: Thread 'SortMerger spilling thread'
terminated due to an exception: The channel is erroneous.
at
org.apache.flink.runtime.operators.sort.UnilateralSortMerger$ThreadBase.run(UnilateralSortMerger.java:801)
Caused by: java.io.IOException: The channel is erroneous.
at
org.apache.flink.runtime.io.disk.iomanager.ChannelAccess.checkErroneous(ChannelAccess.java:132)
at
org.apache.flink.runtime.io.disk.iomanager.BlockChannelWriter.writeBlock(BlockChannelWriter.java:73)
at
org.apache.flink.runtime.io.disk.iomanager.ChannelWriterOutputView.writeSegment(ChannelWriterOutputView.java:220)
at
org.apache.flink.runtime.io.disk.iomanager.ChannelWriterOutputView.nextSegment(ChannelWriterOutputView.java:206)
at
org.apache.flink.runtime.memorymanager.AbstractPagedOutputView.advance(AbstractPagedOutputView.java:140)
at
org.apache.flink.runtime.memorymanager.AbstractPagedOutputView.writeInt(AbstractPagedOutputView.java:267)
at
org.apache.flink.api.common.typeutils.base.IntSerializer.copy(IntSerializer.java:74)
at
org.apache.flink.api.java.typeutils.runtime.TupleSerializer.copy(TupleSerializer.java:124)
at
org.apache.flink.runtime.operators.sort.NormalizedKeySorter.writeToOutput(NormalizedKeySorter.java:449)
at
org.apache.flink.runtime.operators.sort.UnilateralSortMerger$SpillingThread.go(UnilateralSortMerger.java:1316)
at
org.apache.flink.runtime.operators.sort.UnilateralSortMerger$ThreadBase.run(UnilateralSortMerger.java:798)
Caused by: java.io.IOException: No space left on device
at sun.nio.ch.FileDispatcherImpl.write0(Native Method)
at sun.nio.ch.FileDispatcherImpl.write(FileDispatcherImpl.java:60)
at sun.nio.ch.IOUtil.writeFromNativeBuffer(IOUtil.java:93)
at sun.nio.ch.IOUtil.write(IOUtil.java:65)
at sun.nio.ch.FileChannelImpl.write(FileChannelImpl.java:205)
at
org.apache.flink.runtime.io.disk.iomanager.SegmentWriteRequest.write(BlockChannelAccess.java:259)
at
org.apache.flink.runtime.io.disk.iomanager.IOManager$WriterThread.run(IOManager.java:644)
at org.apache.flink.client.program.Client.run(Client.java:325)
at org.apache.flink.client.program.Client.run(Client.java:291)
at org.apache.flink.client.program.Client.run(Client.java:285)
at
org.apache.flink.client.program.ContextEnvironment.execute(ContextEnvironment.java:54)
at
org.apache.flink.allIn.StratosphereMultiFlink6Job.main(StratosphereMultiFlink6Job.java:72)
at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
at
sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:39)
at
sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:25)
at java.lang.reflect.Method.invoke(Method.java:597)
at
org.apache.flink.client.program.PackagedProgram.callMainMethod(PackagedProgram.java:389)
at
org.apache.flink.client.program.PackagedProgram.invokeInteractiveModeForExecution(PackagedProgram.java:307)
at org.apache.flink.client.program.Client.run(Client.java:244)
at
org.apache.flink.client.CliFrontend.executeProgram(CliFrontend.java:332)
at org.apache.flink.client.CliFrontend.run(CliFrontend.java:319)
at
org.apache.flink.client.CliFrontend.parseParameters(CliFrontend.java:930)
at org.apache.flink.client.CliFrontend.main(CliFrontend.java:954)
Am 08.10.2014 15:04, schrieb Ufuk Celebi:
Hey Florian,
this is a known issue and the commit, which introduced the problem has been
reverted for the 0.6.1 release. Could you please work with that version?
– Ufuk
On 08 Oct 2014, at 14:34, Florian Hönicke <[email protected]> wrote:
Hi,
I get a runtime error while executing my flink0.6 job.
"An error occurred in the channel"
Could anyone help me?
Greetings
Florian
<out.txt>
package org.apache.flink.allIn;
import java.util.ArrayList;
import java.util.Iterator;
import org.apache.flink.api.java.DataSet;
import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.api.java.aggregation.Aggregations;
import org.apache.flink.api.java.functions.FunctionAnnotation.ConstantFields;
import org.apache.flink.api.common.functions.GroupReduceFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.api.java.tuple.Tuple4;
import org.apache.flink.api.java.tuple.Tuple7;
import org.apache.flink.util.Collector;
import org.apache.flink.core.fs.FileSystem.WriteMode;
@SuppressWarnings("serial")
public class StratosphereMultiFlink6Job{
public static void main(String[] args) throws Exception {
// set up the execution environment
final ExecutionEnvironment env =
ExecutionEnvironment.getExecutionEnvironment();
DataSet<Tuple3<Integer, Integer, Integer>> InputRecs =
env.readTextFile(args[0])
.map(new SourceMapper());
DataSet<Tuple4<Integer, Integer, Integer, Integer>> Uni =
InputRecs.map(new UniMapper())
.groupBy(0)
.aggregate(Aggregations.SUM, 1)
.joinWithHuge(InputRecs).where(0).equalTo(0)
.projectSecond(0,1,2).projectFirst(1)
.types(Integer.class, Integer.class,
Integer.class, Integer.class)
;
DataSet<Tuple3<Integer, Integer, Double>> sim =
Uni
// .joinWithHuge(Uni).where(0).equalTo(0).with(new
JoinCondition())
// .groupBy(0,1)
// .reduceGroup(new ReduceSim2());
//
.groupBy(1)
.reduceGroup(new ReduceSim1())
.groupBy(0,1)
.reduceGroup(new ReduceSim2());
////////////////////////////////////////////////////////////////////
////////////////////////////////////////////////////////////////////
////////////////////////////////////////////////////////////////////
sim.writeAsText(args[1], WriteMode.OVERWRITE);
// emit result
//sim.print();
// execute program
env.execute("WordCount Example");
}
// public class JoinCondition
// extends JoinFunction<Tuple4<Integer, String, Integer, Integer>,
Tuple4<Integer, String, Integer, Integer>
// , Tuple7<Integer,Integer,Integer,Integer, String, Integer, Integer>> {
//
// @Override
// public Tuple7<Integer,Integer,Integer,Integer, String, Integer,
Integer> join(Tuple4<Integer, String, Integer, Integer> tup1, Tuple4<Integer,
String, Integer, Integer> tup2) {
// // multiply the points and rating and construct a new
output tuple
// if(tup1.f0.intValue()<tup2.f0.intValue()){
// return new
Tuple7<Integer,Integer,Integer,Integer,String,Integer,Integer>(
//
tup1.f0.intValue(),tup2.f0.intValue(),tup1.f3.intValue(),tup2.f3.intValue(),String.valueOf(tup1.f1),tup1.f2.intValue(),tup2.f2.intValue());//i
j//uni i//uni j// ak// fik// fjk
// }else return null;
// }
// }
//
// User Functions
//
/**
* Implements the string tokenizer that splits sentences into words as
a user-defined
* FlatMapFunction. The function takes a line (String) and splits it
into
* multiple pairs in the form of "(word,1)" (Tuple2<String, Integer>).
*/
public static class SourceMapper implements
MapFunction<String,Tuple3<Integer,Integer,Integer>> {
@Override
public Tuple3<Integer,Integer,Integer> map (String in) {
String[] tuple = in.split(" ");
return (new Tuple3<Integer, Integer,
Integer>(Integer.parseInt(tuple[0]),Integer.parseInt(tuple[1]),Integer.parseInt(tuple[2])));
}
}
@ConstantFields("0->0")
public static class UniMapper implements
MapFunction<Tuple3<Integer,Integer,Integer>,Tuple2<Integer,Integer>> {
@Override
public Tuple2<Integer,Integer> map
(Tuple3<Integer,Integer,Integer> in) {
return (new
Tuple2<Integer,Integer>(in.f0.intValue(),in.f2.intValue()));
}
}
// @ConstantFields("1->4")
public static class ReduceSim1 implements
GroupReduceFunction<Tuple4<Integer,Integer,Integer,Integer>,Tuple7<Integer,Integer,Integer,Integer,Integer,Integer,Integer>>
{
@Override
public void
reduce(Iterable<Tuple4<Integer,Integer,Integer,Integer>> in,
Collector<Tuple7<Integer,Integer,Integer,Integer,Integer,Integer,Integer>> out){
ArrayList<Tuple4<Integer,Integer,Integer,Integer>>
tempRecList = new ArrayList<Tuple4<Integer,Integer,Integer,Integer>>();
for (Tuple4<Integer,Integer,Integer,Integer> tup1:in)
{// TODO warum sind hier mehrmals die Identischen Tupel drin?
//if(tempRecList.contains(rec1)){continue;}
//print tempRecList
//p(rec1);
//p(tempRecList.toString());
for(Tuple4<Integer,Integer,Integer,Integer>
tup2: tempRecList){// input recs sind: i, ak, fik, uni
if(tup1.f0<tup2.f0){
out.collect(new
Tuple7<Integer,Integer,Integer,Integer,Integer,Integer,Integer>(
tup1.f0.intValue(),tup2.f0.intValue(),tup1.f3.intValue(),tup2.f3.intValue(),tup1.f1.intValue(),tup1.f2.intValue(),tup2.f2.intValue()));//i
j//uni i//uni j// ak// fik// fjk
}else{
out.collect(new
Tuple7<Integer,Integer,Integer,Integer,Integer,Integer,Integer>(
tup2.f0.intValue(),tup1.f0.intValue(),tup2.f3.intValue(),tup1.f3.intValue(),tup2.f1.intValue(),tup2.f2.intValue(),tup1.f2.intValue()));//i
j//uni i//uni j// ak// fik// fjk
}
}
tempRecList.add(new
Tuple4<Integer,Integer,Integer,Integer>(
tup1.f0.intValue(),
tup1.f1.intValue(),
tup1.f2.intValue(),
tup1.f3.intValue()));
}
tempRecList.clear();
//p("ReduceErgebnis: " + current.getField(0,
IntValue.class)+" "+ count);
}
}
public static class ReduceSim2 implements
GroupReduceFunction<Tuple7<Integer,Integer,Integer,Integer,Integer,Integer,Integer>,Tuple3<Integer,Integer,Double>>
{
@Override
public void
reduce(Iterable<Tuple7<Integer,Integer,Integer,Integer,Integer,Integer,Integer>>
in, Collector<Tuple3<Integer,Integer,Double>> out) {
double sim = 0;
int unii = 0;
int unij = 0;
Iterator<Tuple7<Integer,Integer,Integer,Integer,Integer,Integer,Integer>> it =
in.iterator();
Tuple7<Integer,Integer,Integer,Integer,Integer,Integer,Integer> current = null;
//unilateral partial results (|*|) for Ruzicka sim
double sum = 0;
while (it.hasNext()) {
current = it.next();
sum += Math.min(current.f5, current.f6);
}
unii = current.f2;
unij = current.f3;
sim = sum/(unii+unij-sum);
//p("sim: "+sim+", unii: "+unii+", unij: "+unij+", sum:
"+sum);
out.collect(new Tuple3<Integer,Integer,Double>(
current.f0.intValue(),
current.f1.intValue(),
sim));
// p("SimErgebnis: " + current.f0+" "
// + "" + current.f1+" "
// + "" + Double.toString(sim));
}
}
// public static void p(Object o){
// System.out.println(o);
// };
}##############################################################################$
# Licensed to the Apache Software Foundation (ASF) under one
# or more contributor license agreements. See the NOTICE file
# distributed with this work for additional information
# regarding copyright ownership. The ASF licenses this file
# to you under the Apache License, Version 2.0 (the
# "License"); you may not use this file except in compliance
# with the License. You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
##############################################################################$
#==============================================================================
# Common
#==============================================================================
jobmanager.rpc.address: dbis31
jobmanager.rpc.port: 6124
jobmanager.heap.mb: 20000
taskmanager.heap.mb: 20000
taskmanager.numberOfTaskSlots: 6
parallelization.degree.default: 1
#==============================================================================
# Web Frontend
#==============================================================================
jobmanager.web.port: 8081
webclient.port: 8080
#==============================================================================
# Advanced
#==============================================================================
# The number of buffers for the network stack.
#
taskmanager.network.numberOfBuffers: 40480
# Directories for temporary files.
#
# Add a delimited list for multiple directories, using the system directory
# delimiter (colon ':' on unix) or a comma, e.g.:
# /data1/tmp:/data2/tmp:/data3/tmp
#
# Note: Each directory entry is read from and written to by a different I/O
# thread. You can include the same directory multiple times in order to create
# multiple I/O threads against that directory. This is for example relevant for
# high-throughput RAIDs.
#
# If not specified, the system-specific Java temporary directory (java.io.tmpd$
# property) is taken.
#
# taskmanager.tmp.dirs: /tmp
# Path to the Hadoop configuration directory.
#
# This configuration is used when writing into HDFS. Unless specified otherwis$
# HDFS file creation will use HDFS default settings with respect to block-size,
# replication factor, etc.
#
# You can also directly specify the paths to hdfs-default.xml and hdfs-site.xml
# via keys 'fs.hdfs.hdfsdefault' and 'fs.hdfs.hdfssite'.
#
# fs.hdfs.hadoopconf: /path/to/hadoop/conf/
