D-Net

package eu.dnetlib.data.mapreduce.hbase.lodExport.build;

import org.apache.hadoop.fs.FSDataOutputStream;

import org.apache.hadoop.fs.FileSystem;

import org.apache.hadoop.fs.Path;

import org.apache.hadoop.io.NullWritable;

import org.apache.hadoop.io.Text;

import org.apache.hadoop.io.compress.CompressionCodec;

import org.apache.hadoop.io.compress.GzipCodec;

import org.apache.hadoop.mapred.FileOutputFormat;

import org.apache.hadoop.mapred.JobConf;

import org.apache.hadoop.mapred.RecordWriter;

import org.apache.hadoop.mapred.Reporter;

import org.apache.hadoop.mapred.TextOutputFormat;

import org.apache.hadoop.util.Progressable;

import org.apache.hadoop.util.ReflectionUtils;

import org.apache.log4j.Logger;

import java.io.DataOutputStream;

import java.io.IOException;

public class StreamingTextOutputFormat<K, V> extends TextOutputFormat<K, V> {

    protected static class StreamingLineRecordWriter<K, V> implements RecordWriter<K, V> {

        private static final String newline = "\n";

        private Logger log = Logger.getLogger(this.getClass());

        protected DataOutputStream out;

        private final String keyValueSeparator = "\t";

        private final String valueDelimiter = ",";

        private boolean dataWritten = false;

        public StreamingLineRecordWriter(DataOutputStream out) {

            this.out = out;

        }

        private void writeObject(Object o) throws IOException {

            if (o instanceof Text) {

                Text to = (Text) o;

                //out.write(to.getBytes(), 0, to.getLength());

                //          out.write("AAAAAAAAAA".getBytes(), 0, to.getLength());

                out.writeUTF(to.toString().replace("\n", ""));

            } else {

                out.writeUTF(o.toString());

            }

        }

        public synchronized void write(K key, V value) throws IOException {

            boolean nullKey = key == null || key instanceof NullWritable;

            boolean nullValue = value == null || value instanceof NullWritable;

            if (nullKey && nullValue) {

                return;

            }

            if (!nullKey) {

                // if we've written data before, append a new line

                if (dataWritten) {

                    out.writeUTF(newline);

                }

                // write out the key and separator

                writeObject(key);

                out.writeUTF(keyValueSeparator);

            } else if (!nullValue) {

                // write out the value delimiter

                out.writeUTF(valueDelimiter);

                // write out the value

                writeObject(value);

            }

            // track that we've written some data

            dataWritten = true;

        }

        public synchronized void close(Reporter reporter) throws IOException {

            // if we've written out any data, append a closing newline

            if (dataWritten) {

                out.writeUTF(newline);

            }

            out.close();

        }

    }

    @Override

    public RecordWriter<K, V> getRecordWriter(FileSystem fileSystem, JobConf job, String name, Progressable progress) throws IOException {

        boolean isCompressed = getCompressOutput(job);

        if (!isCompressed) {

            Path file = FileOutputFormat.getTaskOutputPath(job, name);

            FileSystem fs = file.getFileSystem(job);

            FSDataOutputStream fileOut = fs.create(file, progress);

            return new StreamingLineRecordWriter<K, V>(fileOut);

        } else {

            Class<? extends CompressionCodec> codecClass = getOutputCompressorClass(

                    job, GzipCodec.class);

            // create the named codec

            CompressionCodec codec = ReflectionUtils.newInstance(codecClass,

                    job);

            // build the filename including the extension

            Path file = FileOutputFormat.getTaskOutputPath(job,

                    name + codec.getDefaultExtension());

            FileSystem fs = file.getFileSystem(job);

            FSDataOutputStream fileOut = fs.create(file, progress);

            return new StreamingLineRecordWriter<K, V>(new DataOutputStream(codec.createOutputStream(fileOut)));

        }

    }

}