
// BEGIN_COPYRIGHT
// 
// Copyright 2009-2015 CRS4.
// 
// 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 at
// 
//   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.
// 
// END_COPYRIGHT

package it.crs4.pydoop;

import java.io.DataOutputStream;
import java.io.IOException;

import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.FSDataOutputStream;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.compress.CompressionCodec;
import org.apache.hadoop.io.compress.GzipCodec;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapred.JobConf;
import org.apache.hadoop.mapred.RecordWriter;
import org.apache.hadoop.mapred.TextOutputFormat;
import org.apache.hadoop.util.*;
import org.apache.hadoop.util.Progressable;

/**
 * A TextOutputFormat that doesn't insert a separator between key and value.
 */
public class NoSeparatorTextOutputFormat extends TextOutputFormat<Text, Text>
{
	 public RecordWriter<Text, Text> getRecordWriter(FileSystem ignored,
                                                  JobConf job,
                                                  String name,
                                                  Progressable progress)
    throws IOException {
    boolean isCompressed = getCompressOutput(job);
    String keyValueSeparator = "";
    if (!isCompressed) {
      Path file = TextOutputFormat.getTaskOutputPath(job, name);
      FileSystem fs = file.getFileSystem(job);
      FSDataOutputStream fileOut = fs.create(file, progress);
      return new LineRecordWriter<Text, Text>(fileOut, keyValueSeparator);
    } 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 = 
        TextOutputFormat.getTaskOutputPath(job, 
                                           name + codec.getDefaultExtension());
      FileSystem fs = file.getFileSystem(job);
      FSDataOutputStream fileOut = fs.create(file, progress);
      return new LineRecordWriter<Text, Text>(new DataOutputStream
                                        (codec.createOutputStream(fileOut)),
                                        keyValueSeparator);
    }
  }
}

@@ -714,16 +714,61 @@
 ption;%0A%0A
+import org.apache.hadoop.conf.Configuration;%0A
 import o
@@ -1053,16 +1053,24 @@
 pred
-.JobConf
+uce.RecordWriter
 ;%0Aim
@@ -1098,29 +1098,38 @@
 p.mapred
-.RecordWriter
+uce.TaskAttemptContext
 ;%0Aimport
@@ -1153,16 +1153,30 @@
 p.mapred
+uce.lib.output
 .TextOut
@@ -1433,9 +1433,9 @@
 %3E%0A%7B%0A
-%09
+ 
  pub
@@ -1463,16 +1463,26 @@
 , Text%3E 
+%0A         
 getRecor
@@ -1493,90 +1493,30 @@
 ter(
-FileSystem ignored,%0A                                                  JobConf
+TaskAttemptContext
  job
-,
 %0A   
@@ -1541,514 +1541,277 @@
     
-                         String name,%0A                                                  Progressable progress)%0A    throws IOException %7B%0A    boolean isCompressed = getCompressOutput(job);%0A    String keyValueSeparator = %22%22;%0A    if (!isCompressed) %7B%0A      Path file = TextOutputFormat.getTaskOutputPath(job, name);%0A      FileSystem fs = file.getFileSystem(job);%0A      FSDataOutputStream fileOut = fs.create(file, progress);%0A      return new LineRecordWriter%3CText, Text%3E(fileOut, keyValueSeparator);%0A    %7D else
+) throws IOException, InterruptedException %7B%0A    final String keyValueSeparator = %22%22;%0A    Configuration conf = job.getConfiguration();%0A    boolean isCompressed = getCompressOutput(job);%0A    CompressionCodec codec = null;%0A    String extension = %22%22;%0A    if (isCompressed)
  %7B%0A 
@@ -1861,16 +1861,17 @@
 cClass =
+ 
 %0A       
@@ -1929,40 +1929,17 @@
     
-// create the named codec%0A      
+codec = (
 Comp
@@ -1950,24 +1950,17 @@
 ionCodec
- codec =
+)
  Reflect
@@ -1992,19 +1992,20 @@
 cClass, 
-job
+conf
 );%0A     
@@ -2009,53 +2009,45 @@
     
-// build the filename including the e
+extension = codec.getDefaultE
 xtension
 %0A   
@@ -2042,19 +2042,26 @@
 xtension
+();
 %0A  
+  %7D%0A
     Path
@@ -2072,133 +2072,41 @@
 e = 
-%0A        TextOutputFormat.getTaskOutputPath(job, %0A                                           name + codec.getDefaultE
+getDefaultWorkFile(job, e
 xtension
 ());
@@ -2105,15 +2105,11 @@
 sion
-()
 );%0A
-  
     
@@ -2143,22 +2143,48 @@
 eSystem(
-job
+conf
 );%0A
+    if (!isCompressed) %7B%0A
       FS
@@ -2230,16 +2230,13 @@
 le, 
-progress
+false
 );%0A 
@@ -2284,145 +2284,223 @@
 xt%3E(
-new DataOutputStream%0A                                        (codec.createOutputStream(fileOut)),%0A                                       
+fileOut, keyValueSeparator);%0A    %7D else %7B%0A      FSDataOutputStream fileOut = fs.create(file, false);%0A      return new LineRecordWriter%3CText, Text%3E(%0A          new DataOutputStream (codec.createOutputStream(fileOut)),
  key

Update NoSeparatorTextOutputFormat for Hadoop 2