Re: Union-Method does not function

Mon, 18 Aug 2014 11:56:39 -0700 

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());
    }
}

Reply via email to