Is copying flink-hadoop-compatibility jar to FLINK_HOME/lib the only way to make it work?

[morven huang]

Hi,

 

I’m using Flink 1.5.6 and Hadoop 2.7.1.

 

My requirement is to read hdfs sequence file (SequenceFileInputFormat), then write it back to hdfs (SequenceFileAsBinaryOutputFormat with compression).

 

Below code won’t work until I copy the flink-hadoop-compatibility jar to FLINK_HOME/lib. I find a similar discussion http://apache-flink-user-mailing-list-archive.2336050.n4.nabble.com/hadoopcompatibility-not-in-dist-td12552.html, it seems this is the only way to get the hadoop compatibility work?

 

If this is the case, do I need to copy that jar to every node of the cluster?

 

Or, for my super simple requirement above, is there any other way to go? For example, without using  flink-hadoop-compatibility?

 

import org.apache.flink.api.common.functions.FlatMapFunction;

import org.apache.flink.api.common.typeinfo.TypeInformation;

import org.apache.flink.api.java.ExecutionEnvironment;

import org.apache.flink.api.java.hadoop.mapreduce.HadoopOutputFormat;

import org.apache.flink.api.java.operators.DataSource;

import org.apache.flink.api.java.operators.FlatMapOperator;

import org.apache.flink.api.java.tuple.Tuple2;

import org.apache.flink.api.java.typeutils.TupleTypeInfo;

import org.apache.flink.hadoopcompatibility.HadoopInputs;

import org.apache.flink.util.Collector;

import org.apache.hadoop.fs.Path;

import org.apache.hadoop.io.BytesWritable;

import org.apache.hadoop.io.NullWritable;

import org.apache.hadoop.io.SequenceFile.CompressionType;

import org.apache.hadoop.mapreduce.Job;

import org.apache.hadoop.mapreduce.lib.output.SequenceFileAsBinaryOutputFormat;

import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;

 

import com.twitter.chill.protobuf.ProtobufSerializer;

 

public class Foobar {

 

        @SuppressWarnings("serial")

        public static void main(String[] args) throws Exception {

                 ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();

                env.getConfig().registerTypeWithKryoSerializer(ProtobufObject.class, ProtobufSerializer.class);

 

                 String path = "hdfs://...";

                 DataSource<Tuple2<NullWritable, BytesWritable>> input = env.createInput(HadoopInputs.readHadoopFile(

                                  new org.apache.hadoop.mapreduce.lib.input.SequenceFileInputFormat<NullWritable, BytesWritable>(),

                                  NullWritable.class, BytesWritable.class, path),

                                  new TupleTypeInfo<>(TypeInformation.of(NullWritable.class), TypeInformation.of(BytesWritable.class)));

 

                 FlatMapOperator<Tuple2<NullWritable, BytesWritable>, Tuple2<BytesWritable, BytesWritable>> x = input.flatMap(

                                  new FlatMapFunction<Tuple2<NullWritable, BytesWritable>, Tuple2<BytesWritable, BytesWritable>>() {

 

                                          @Override

                                          public void flatMap(Tuple2<NullWritable, BytesWritable> value,

                                                           Collector<Tuple2<BytesWritable, BytesWritable>> out) throws Exception {

                                                   ProtobufObject info = ProtobufObject.parseFrom(value.f1.copyBytes());

                                                   String key = info.getKey();

                                                   out.collect(new Tuple2<BytesWritable, BytesWritable>(new BytesWritable(key.getBytes()),

                                                                   new BytesWritable(info.toByteArray())));

                                          }

                                  });

 

                 Job job = Job.getInstance();

                 HadoopOutputFormat<BytesWritable, BytesWritable> hadoopOF = new HadoopOutputFormat<BytesWritable, BytesWritable>(

                                  new SequenceFileAsBinaryOutputFormat(), job);

 

                hadoopOF.getConfiguration().set("mapreduce.output.fileoutputformat.compress", "true");

                hadoopOF.getConfiguration().set("mapreduce.output.fileoutputformat.compress.type", CompressionType.BLOCK.toString());

                 TextOutputFormat.setOutputPath(job, new Path("hdfs://..."));

 

                 x.output(hadoopOF);

                 env.execute("foo");

        }

}

 

Sent from Mail for Windows 10