@Fabian: What was the source of the error? Can you open a JIRA issue with the description?
On Mon, Aug 18, 2014 at 10:49 PM, Fabian Hueske <[email protected]> wrote: > Hi Alex, > > thanks a lot for posting your program! > I modified the code a little bit and tested it against the current master > without problems (code pasted below). > > We will release a new version that includes many bugfixes very soon. If > you do not want to wait for the release, I suggest to clone the > release-0.6-rc7 branch from the Flink Git repository. > > I also noticed that your code is not doing a proper outer join. Outer > joins preserve all tuples that do not find a matching join partner and fill > the missing attributes with null values. With coGroup you can preserve > tuples without join partners (those where one iterator is empty). > Adding the tuple from the collection set using the union, just adds an > additional tuple that will always join with itself and will be included in > the result. > > Please let me know, if you have further questions or observe any other > problem with the system. > > Best, Fabian > > ---------- Code ---------------- > > public class ProgTest { > > @SuppressWarnings({ "unchecked", "serial" }) > public static void main(String[] args) throws Exception { > ArrayList<Tuple2<Integer, String>> data = new > ArrayList<Tuple2<Integer, String>>(); > data.add(new Tuple2<Integer, String>(1,"Peter")); > data.add(new Tuple2<Integer, String>(2,"Max")); > data.add(new Tuple2<Integer, String>(3,"Sarah")); > data.add(new Tuple2<Integer, String>(4,"Julia")); > > > // set up the execution environment > final ExecutionEnvironment env = ExecutionEnvironment > .getExecutionEnvironment(); > // get input data > DataSet<Tuple3<String, Integer, Integer>> names = > env.fromCollection(data).map( > new MapFunction<Tuple2<Integer,String>, > Tuple3<String,Integer,Integer>>() { > public Tuple3<String,Integer,Integer> > map(Tuple2<Integer,String> line) throws Exception { > return new > Tuple3<String,Integer,Integer>(line.f1, line.f0, 2); > } > }); > DataSet<Tuple3<String, Integer, Integer>> myNull = env > .fromElements(new Tuple3<String, Integer, Integer>("-1", > new Integer(-1), new Integer(-1))); > names = names.union(myNull); > DataSet<Tuple2<String, String>> result = > names.join(names).where(2) > .equalTo(1).projectFirst(0).projectSecond(0) > .types(String.class, String.class); > // emit result > result.writeAsCsv("file:///testOut", "\n", " | "); > // execute program > env.execute(); > } > } > > > > > 2014-08-18 20:54 GMT+02:00 Alexander Sirotin <[email protected]>: > > Hi Fabian, >> >> thank you for your help. I will try the CoGroup-function. >> >> At the end of this email there is my program. I changed it and attach >> therefore the stracktrace again. After the join I called union with the >> myNull-DataSet. This DataSet was created by calling fromElements - so not >> read from a file. The union-function then caused the showed Exception and I >> didnt get any result. Without the union the programs functions, but then it >> is an inner-join. Anyway I will try the CoGroup-method. >> >> Thank you again and best regards, >> Alex >> >> Error: The main method caused an error. >> >> eu.stratosphere.client.program.ProgramInvocationException: The main >> method caused an error. >> at eu.stratosphere.client.program.PackagedProgram.callMainMethod( >> PackagedProgram.java:399) >> at eu.stratosphere.client.program.PackagedProgram. >> invokeInteractiveModeForExecution(PackagedProgram.java:302) >> at eu.stratosphere.client.program.Client.run(Client.java:246) >> at eu.stratosphere.client.CliFrontend.executeProgram( >> CliFrontend.java:327) >> at eu.stratosphere.client.CliFrontend.run(CliFrontend.java:314) >> at eu.stratosphere.client.CliFrontend.parseParameters( >> CliFrontend.java:927) >> at eu.stratosphere.client.CliFrontend.main(CliFrontend.java:951) >> Caused by: eu.stratosphere.compiler.CompilerException: Error translating >> node 'Union "Union" : UNION [[ GlobalProperties >> [partitioning=HASH_PARTITIONED, >> on fields [2]] ]] [[ LocalProperties [ordering=null, grouped=null, >> unique=null] ]]': null >> at eu.stratosphere.compiler.plantranslate. >> NepheleJobGraphGenerator.preVisit(NepheleJobGraphGenerator.java:355) >> at eu.stratosphere.compiler.plantranslate. >> NepheleJobGraphGenerator.preVisit(NepheleJobGraphGenerator.java:99) >> at eu.stratosphere.compiler.plan.DualInputPlanNode.accept( >> DualInputPlanNode.java:157) >> at eu.stratosphere.compiler.plan.DualInputPlanNode.accept( >> DualInputPlanNode.java:159) >> at eu.stratosphere.compiler.plan.SingleInputPlanNode.accept( >> SingleInputPlanNode.java:141) >> at eu.stratosphere.compiler.plan.OptimizedPlan.accept( >> OptimizedPlan.java:159) >> at eu.stratosphere.compiler.plantranslate. >> NepheleJobGraphGenerator.compileJobGraph(NepheleJobGraphGenerator.java: >> 169) >> at eu.stratosphere.client.program.Client.getJobGraph( >> Client.java:222) >> at eu.stratosphere.client.program.Client.run(Client.java:293) >> at eu.stratosphere.client.program.Client.run(Client.java:288) >> at eu.stratosphere.client.program.ContextEnvironment. >> execute(ContextEnvironment.java:50) >> at de.hshannover.vis.flink.jobdb.A3NameWithBoss.main( >> A3NameWithBoss.java:59) >> >> at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method) >> at sun.reflect.NativeMethodAccessorImpl.invoke( >> NativeMethodAccessorImpl.java:57) >> at sun.reflect.DelegatingMethodAccessorImpl.invoke( >> DelegatingMethodAccessorImpl.java:43) >> at java.lang.reflect.Method.invoke(Method.java:606) >> at eu.stratosphere.client.program.PackagedProgram.callMainMethod( >> PackagedProgram.java:384) >> ... 6 more >> Caused by: java.lang.NullPointerException >> at eu.stratosphere.pact.runtime.task.util.TaskConfig. >> setDriver(TaskConfig.java:291) >> at eu.stratosphere.compiler.plantranslate. >> NepheleJobGraphGenerator.createDualInputVertex( >> NepheleJobGraphGenerator.java:812) >> at eu.stratosphere.compiler.plantranslate. >> NepheleJobGraphGenerator.preVisit(NepheleJobGraphGenerator.java:303) >> ... 22 more >> >> >> package de.hshannover.vis.flink.jobdb; >> >> import java.util.ArrayList; >> import java.util.List; >> >> import de.hshannover.vis.flink.jobdb.mapper.EmployeeMapper; >> import de.hshannover.vis.flink.jobdb.utils.IParameter; >> import de.hshannover.vis.flink.jobdb.utils.Loader; >> import de.hshannover.vis.flink.jobdb.utils.OutputPath; >> import eu.stratosphere.api.java.DataSet; >> import eu.stratosphere.api.java.ExecutionEnvironment; >> import eu.stratosphere.api.java.functions.*; >> import eu.stratosphere.api.java.tuple.*; >> >> public class A3NameWithBoss { >> >> @SuppressWarnings({ "unchecked", "serial" }) >> public static void main(String[] args) throws Exception { >> >> List<IParameter> list = new ArrayList<IParameter>(); >> list.add(new EmployeeMapper()); >> list.add(new OutputPath("Output", "")); >> Loader loader = new Loader(list, args); >> >> loader.load(); >> >> // set up the execution environment >> final ExecutionEnvironment env = ExecutionEnvironment >> .getExecutionEnvironment(); >> >> // get input data >> DataSet<Tuple3<String, Integer, Integer>> names = >> loader.data.get(0).map( >> new MapFunction<Tuple11 >> <Integer,String,String,String,String,String,String,Integer, >> Double,Integer,Integer> >> , Tuple3<String,Integer,Integer>>() { >> public Tuple3<String,Integer,Integer> map(Tuple11 >> <Integer,String,String,String,String,String,String,Integer, >> Double,Integer,Integer> >> line) throws Exception { >> return new Tuple3<String,Integer,Integer>(line.f1 >> + " " + line.f2, line.f0, >> line.f9); >> } >> >> }); >> >> >> DataSet<Tuple3<String, Integer, Integer>> myNull = env >> .fromElements(new Tuple3<String, Integer, Integer>("-1", >> new Integer(-1), new Integer(-1))); >> >> names = names.union(myNull); >> >> DataSet<Tuple2<String, String>> result = >> names.join(names).where(2) >> .equalTo(1).projectFirst(0).projectSecond(0) >> .types(String.class, String.class); >> >> // emit result >> result.writeAsCsv(loader.output.get(0).getPath(), "\n", " | "); >> >> // execute program >> env.execute(A3NameWithBoss.class.toString()); >> } >> } >> > >
