<commit_before>/*
 * Copyright 2009-2010 by The Regents of the University of California
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * you may obtain a copy of the License from
 * 
 *     http://www.apache.org/licenses/LICENSE-2.0
 * 
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package edu.uci.ics.hyracks.dataflow.hadoop;

import java.io.File;
import java.io.IOException;
import java.io.OutputStream;
import java.util.HashMap;
import java.util.Map;

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.SequenceFile;
import org.apache.hadoop.io.SequenceFile.Writer;
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.InputSplit;
import org.apache.hadoop.mapred.JobConf;
import org.apache.hadoop.mapred.Reporter;
import org.apache.hadoop.mapred.SequenceFileOutputFormat;
import org.apache.hadoop.mapred.Counters.Counter;
import org.apache.hadoop.mapred.lib.NullOutputFormat;
import org.apache.hadoop.mapreduce.JobContext;
import org.apache.hadoop.mapreduce.TaskAttemptContext;
import org.apache.hadoop.mapreduce.TaskAttemptID;
import org.apache.hadoop.util.Progressable;
import org.apache.hadoop.util.ReflectionUtils;

import edu.uci.ics.hyracks.api.exceptions.HyracksException;
import edu.uci.ics.hyracks.api.job.JobSpecification;
import edu.uci.ics.hyracks.dataflow.hadoop.util.DatatypeHelper;
import edu.uci.ics.hyracks.dataflow.std.file.AbstractFileWriteOperatorDescriptor;
import edu.uci.ics.hyracks.dataflow.std.file.FileSplit;
import edu.uci.ics.hyracks.dataflow.std.file.IRecordWriter;
import edu.uci.ics.hyracks.dataflow.std.file.RecordWriter;

public class HadoopWriteOperatorDescriptor extends AbstractFileWriteOperatorDescriptor {

    private static String nullWritableClassName = NullWritable.class.getName();

    private  class HadoopFileWriter implements IRecordWriter {

        Object recordWriter;
        JobConf conf;
        final boolean useNewMapReduceLib;
        Path finalOutputFile;
        Path tempOutputFile;
      
        
        HadoopFileWriter(Object recordWriter,Path tempOutputFile,Path outputFile,JobConf conf) {
            this.recordWriter = recordWriter;
            this.conf = conf;
            this.finalOutputFile = outputFile;
            this.tempOutputFile = tempOutputFile;
            useNewMapReduceLib = conf.getUseNewMapper();
        }

        @Override
        public void write(Object[] record) throws Exception {
            if (useNewMapReduceLib){
                ((org.apache.hadoop.mapreduce.RecordWriter)recordWriter).write(record[0], record[1]);
            } else {
                ((org.apache.hadoop.mapred.RecordWriter)recordWriter).write(record[0], record[1]);
            }    
        }

        @Override
        public void close() {
            try {
                if (useNewMapReduceLib){
                   ((org.apache.hadoop.mapreduce.RecordWriter)recordWriter).close(new TaskAttemptContext(conf, new TaskAttemptID()));
                } else {
                    ((org.apache.hadoop.mapred.RecordWriter)recordWriter).close(null);
                    FileSystem.get(conf).rename( tempOutputFile, finalOutputFile);
                }
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
    }

    private static class HadoopSequenceWriter implements IRecordWriter {
        private Writer writer;

        HadoopSequenceWriter(Writer writer) throws Exception {
            this.writer = writer;
        }

        @Override
        public void close() {
            try {
                writer.close();
            } catch (IOException e) {
                e.printStackTrace();
            }
        }

        @Override
        public void write(Object[] record) throws Exception {
            Object key = record[0];
            Object value = record[1];
            writer.append(key, value);
        }
    }

    private static final long serialVersionUID = 1L;
    Map<String, String> jobConfMap;

    @Override
    protected IRecordWriter createRecordWriter(FileSplit fileSplit, int index) throws Exception {
        JobConf conf = DatatypeHelper.map2JobConf((HashMap) jobConfMap);
        conf.setClassLoader(this.getClass().getClassLoader());
        Thread.currentThread().setContextClassLoader(this.getClass().getClassLoader());
        FileSystem fileSystem = null;
        try {
            fileSystem = FileSystem.get(conf);
        } catch (IOException ioe) {
            ioe.printStackTrace();
        }
        Path path = new Path(fileSplit.getPath());
        Path tempOutputFile = null;
        Path finalOutputFile = null;
        checkIfCanWriteToHDFS(new FileSplit[] { fileSplit });
        Object recordWriter  = null;
        boolean sequenceFileOutput = false;
        Object outputFormat = null;
        if(conf.getUseNewMapper()){
            org.apache.hadoop.mapreduce.OutputFormat newOutputFormat = (org.apache.hadoop.mapreduce.OutputFormat)ReflectionUtils.newInstance((new JobContext(conf,null)).getOutputFormatClass(),conf);
            if(newOutputFormat instanceof org.apache.hadoop.mapreduce.lib.output.SequenceFileOutputFormat) {
                 sequenceFileOutput = true;
            }else{
                recordWriter = newOutputFormat.getRecordWriter(new TaskAttemptContext(conf, new TaskAttemptID()));
            }    
        }else {
            if(conf.getOutputFormat() instanceof SequenceFileOutputFormat) {
                sequenceFileOutput = true;
            } else {
                String taskAttempId = new TaskAttemptID().toString();
                conf.set("mapred.task.id",taskAttempId);
                outputPath = new Path(conf.get("mapred.output.dir"));
                outputTempPath = new Path(outputPath,"_temporary");
                if(outputPath != null && !fileSystem.exists(outputPath)) {
                    fileSystem.mkdirs(outputTempPath);
                }
                String suffix = new String("part-00000");
                suffix = new String(suffix.substring(0, suffix.length() - ("" + index).length()));
                suffix = suffix + index;
                tempOutputFile = new Path(outputTempPath,"_" + taskAttempId + "/" + suffix);
                finalOutputFile = new Path(outputPath,suffix);
                recordWriter = conf.getOutputFormat().getRecordWriter(fileSystem, conf,suffix, new Progressable() {
                @Override
                public void progress() {}
                });
            }
        }
        if(!sequenceFileOutput) {
            return new HadoopFileWriter(recordWriter,tempOutputFile,finalOutputFile,conf);
        } else { 
            Class keyClass = Class.forName(conf.getOutputKeyClass().getName());
            Class valueClass = Class.forName(conf.getOutputValueClass().getName());
            conf.setClass("mapred.output.compression.codec", GzipCodec.class, CompressionCodec.class);
            Writer writer = SequenceFile.createWriter(fileSystem, conf, path, keyClass, valueClass);
            return new HadoopSequenceWriter(writer);    
        }   
    }
    

    Path outputPath;
    Path outputTempPath;
   
    protected Reporter createReporter() {
    return new Reporter() {
        @Override
        public Counter getCounter(Enum<?> name) {<commit_msg>Modified HadoopWriterOperatorDescriptor. The operator previously used to create Sequence files by opening FSDataOutputStream. Though this results in correct creation of the sequence file,  it is still better to have it done by obtaining a SequenceFile writer from the outputFormat.<commit_after>/*
 * Copyright 2009-2010 by The Regents of the University of California
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * you may obtain a copy of the License from
 * 
 *     http://www.apache.org/licenses/LICENSE-2.0
 * 
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package edu.uci.ics.hyracks.dataflow.hadoop;

import java.io.File;
import java.io.IOException;
import java.util.HashMap;
import java.util.Map;

import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.SequenceFile.Writer;
import org.apache.hadoop.mapred.FileOutputFormat;
import org.apache.hadoop.mapred.InputSplit;
import org.apache.hadoop.mapred.JobConf;
import org.apache.hadoop.mapred.Reporter;
import org.apache.hadoop.mapred.SequenceFileOutputFormat;
import org.apache.hadoop.mapred.Counters.Counter;
import org.apache.hadoop.mapred.lib.NullOutputFormat;
import org.apache.hadoop.mapreduce.JobContext;
import org.apache.hadoop.mapreduce.TaskAttemptContext;
import org.apache.hadoop.mapreduce.TaskAttemptID;
import org.apache.hadoop.util.Progressable;
import org.apache.hadoop.util.ReflectionUtils;

import edu.uci.ics.hyracks.api.job.JobSpecification;
import edu.uci.ics.hyracks.dataflow.hadoop.util.DatatypeHelper;
import edu.uci.ics.hyracks.dataflow.std.file.AbstractFileWriteOperatorDescriptor;
import edu.uci.ics.hyracks.dataflow.std.file.FileSplit;
import edu.uci.ics.hyracks.dataflow.std.file.IRecordWriter;

public class HadoopWriteOperatorDescriptor extends AbstractFileWriteOperatorDescriptor {

    private  class HadoopFileWriter implements IRecordWriter {

        Object recordWriter;
        JobConf conf;
        Path finalOutputFile;
        Path tempOutputFile;
      
        
        HadoopFileWriter(Object recordWriter,Path tempOutputFile,Path outputFile,JobConf conf) {
            this.recordWriter = recordWriter;
            this.conf = conf;
            this.finalOutputFile = outputFile;
            this.tempOutputFile = tempOutputFile;
        }

        @Override
        public void write(Object[] record) throws Exception {
            if (conf.getUseNewMapper()){
                ((org.apache.hadoop.mapreduce.RecordWriter)recordWriter).write(record[0], record[1]);
            } else {
                ((org.apache.hadoop.mapred.RecordWriter)recordWriter).write(record[0], record[1]);
            }    
        }

        @Override
        public void close() {
            try {
                if (conf.getUseNewMapper()){
                   ((org.apache.hadoop.mapreduce.RecordWriter)recordWriter).close(new TaskAttemptContext(conf, new TaskAttemptID()));
                } else {
                    ((org.apache.hadoop.mapred.RecordWriter)recordWriter).close(null);
                }
                FileSystem.get(conf).rename( tempOutputFile, finalOutputFile);
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
    }

    private static class HadoopSequenceWriter implements IRecordWriter {
        private Writer writer;

        HadoopSequenceWriter(Writer writer) throws Exception {
            this.writer = writer;
        }

        @Override
        public void close() {
            try {
                writer.close();
            } catch (IOException e) {
                e.printStackTrace();
            }
        }

        @Override
        public void write(Object[] record) throws Exception {
            Object key = record[0];
            Object value = record[1];
            writer.append(key, value);
        }
    }

    private static final long serialVersionUID = 1L;
    Map<String, String> jobConfMap;

    @Override
    protected IRecordWriter createRecordWriter(FileSplit fileSplit, int index) throws Exception {
        JobConf conf = DatatypeHelper.map2JobConf((HashMap) jobConfMap);
        conf.setClassLoader(this.getClass().getClassLoader());
        Thread.currentThread().setContextClassLoader(this.getClass().getClassLoader());
        FileSystem fileSystem = null;
        try {
            fileSystem = FileSystem.get(conf);
        } catch (IOException ioe) {
            ioe.printStackTrace();
        }
        Path path = new Path(fileSplit.getPath());
        Path tempOutputFile = null;
        Path finalOutputFile = null;
        checkIfCanWriteToHDFS(new FileSplit[] { fileSplit });
        Object recordWriter  = null;
        Object outputFormat = null;
        String taskAttempId = new TaskAttemptID().toString();
        conf.set("mapred.task.id",taskAttempId);
        outputPath = new Path(conf.get("mapred.output.dir"));
        outputTempPath = new Path(outputPath,"_temporary");
        if(outputPath != null && !fileSystem.exists(outputPath)) {
            fileSystem.mkdirs(outputTempPath);
        }
        String suffix =  new String("part-r-00000");
        suffix = new String(suffix.substring(0, suffix.length() - ("" + index).length()));
        suffix = suffix + index;
        tempOutputFile = new Path(outputTempPath,"_" + taskAttempId + "/" + suffix);
        if (conf.getNumReduceTasks() == 0 ) {
            suffix.replace("-r-", "-m-");
        }
        finalOutputFile = new Path(outputPath,suffix);
        if(conf.getUseNewMapper()){
            org.apache.hadoop.mapreduce.OutputFormat newOutputFormat = (org.apache.hadoop.mapreduce.OutputFormat)ReflectionUtils.newInstance((new JobContext(conf,null)).getOutputFormatClass(),conf);
                recordWriter = newOutputFormat.getRecordWriter(new TaskAttemptContext(conf, new TaskAttemptID()));
        }else {
           recordWriter = conf.getOutputFormat().getRecordWriter(fileSystem, conf,suffix, new Progressable() {
           @Override
           public void progress() {}
           });
        }
        
        
        return new HadoopFileWriter(recordWriter, tempOutputFile, finalOutputFile, conf);
    }
    

    Path outputPath;
    Path outputTempPath;
   
    protected Reporter createReporter() {
    return new Reporter() {
        @Override
        public Counter getCounter(Enum<?> name) {
{'fraction_non_alphanumeric': 0.0970210440010932, 'fraction_numerical': 0.0036212079803224924, 'mean_word_length': 2.6748681898066784, 'pattern_counts': {'":': 0, '<': 7, '<?xml version=': 0, '>': 7, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '6895228', 'n_tokens_mistral': 3946, 'n_tokens_neox': 3821, 'n_words': 850}