Flink - Hadoop Connectivity - Unable to read file

[Samik Mukherjee]

Hi All,

I am trying to get some file from HDFS which is locally installed. But I am
not able to. I tried with both these ways. But all the time the program is
ending with "Process finished with exit code 239." Any help will be helpful-

public class Processor {


  public static void main(String[] args) throws Exception {
        final ExecutionEnvironment env =
ExecutionEnvironment.*getExecutionEnvironment*();
        //env.setParallelism(1);

        Job job = Job.*getInstance*();
        HadoopInputFormat<LongWritable, Text> hadoopInputFormat = new
HadoopInputFormat<LongWritable, Text>(new TextInputFormat(),
LongWritable.class, Text.class, job);
        TextInputFormat.*addInputPath*(job, new
Path("hdfs://localhost:9000/newfolder/testdata1"));
        DataSet<Tuple2<LongWritable, Text>> text =
env.createInput(hadoopInputFormat);
        DataSet<Tuple2<String, Integer>> words = text.flatMap(new Tokenizer());



        words.print();
        env.execute("Processor");
    }


    public static final class Tokenizer extends
RichFlatMapFunction<Tuple2<LongWritable, Text>, Tuple2<String,
Integer>> {
        @Override
        public void flatMap(Tuple2<LongWritable, Text> value,
org.apache.flink.util.Collector<Tuple2<String, Integer>> out) throws
Exception {
            // normalize and split the line
            String line = value.f1.toString();
            String[] tokens = line.toLowerCase().split("\\W+");

            // emit the pairs
            for (String token : tokens) {
                if (token.length() > 0) {
                    out.collect(new Tuple2<String, Integer>(token, 1));
                }
            }
        }
    }
}








  public static void main(String[] args) throws Exception {
        final ExecutionEnvironment env =
ExecutionEnvironment.*getExecutionEnvironment*();
        //env.setParallelism(1);

        DataSet<Tuple2<LongWritable, Text>> input;

        input = env.createInput(HadoopInputs.*readHadoopFile*(new
TextInputFormat(),
                LongWritable.class, Text.class,
"hdfs://localhost:9000/newfolder/testdata1"));

        DataSet<String> stringInput = input.map(new
MapFunction<Tuple2<LongWritable, Text>, String>() {
            @Override
            public String map(Tuple2<LongWritable, Text>
longWritableTextTuple2) throws Exception {
                return longWritableTextTuple2.f1.toString();
            }
        });

        stringInput.print();

        env.execute("Processor");
    }