Hadoop 流 - 从减速器输出中删除尾随选项卡

我有一个 hadoop 流作业，其输出不包含键/值对。您可以将其视为仅值对或仅键对。

我的流式减速器（一个 php 脚本）正在输出由换行符分隔的记录。 Hadoop 流处理将此视为没有值的键，并在换行符之前插入一个制表符。这个额外的选项卡是不需要的。

我该如何删除它？

我正在将 hadoop 1.0.3 与 AWS EMR 结合使用。我下载了hadoop 1.0.3的源代码，并在hadoop-1.0.3/src/contrib/streaming/src/java/org/apache/hadoop/streaming/PipeReducer.java中找到了这段代码：

reduceOutFieldSeparator = job_.get("stream.reduce.output.field.separator", "\t").getBytes("UTF-8");

所以我尝试通过-D stream.reduce.output.field.separator=作为这项工作的论据，但运气不佳。我也尝试过-D mapred.textoutputformat.separator= and -D mapreduce.output.textoutputformat.separator=没有运气。

我当然搜索过谷歌，但没有找到任何有用的东西。一项搜索结果甚至指出没有可以传递的参数来实现所需的结果（尽管在这种情况下，hadoop 版本确实非常旧）。

这是我的代码（为了可读性添加了换行符）：

hadoop jar streaming.jar -files s3n://path/to/a/file.json#file.json
    -D mapred.output.compress=true -D stream.reduce.output.field.separator=
    -input s3n://path/to/some/input/*/* -output hdfs:///path/to/output/dir
    -mapper 'php my_mapper.php' -reducer 'php my_reducer.php'

为了对其他人有帮助，使用上面的提示，我能够实现：

CustomOutputFormat<K, V> extends org.apache.hadoop.mapred.TextOutputFormat<K, V> {....}

将“getRecordWriter”的内置实现中的一行更改为：

String keyValueSeparator = job.get("mapred.textoutputformat.separator", ""); 

代替：

String keyValueSeparator = job.get("mapred.textoutputformat.separator", "\t"); 

将其编译成 Jar 并将其包含到我的 hadoop 流式调用中（通过 hadoop 流式处理上的说明）后，该调用如下所示：

hadoop   jar  /usr/lib/hadoop/contrib/streaming/hadoop-streaming-1.0.3.jar     \
-archives 'hdfs:///user/the/path/to/your/jar/onHDFS/theNameOfTheJar.jar' \
-libjars theNameOfTheJar.jar \
-outputformat com.yourcompanyHere.package.path.tojavafile.CustomOutputFormat  \
-file yourMapper.py    -mapper  yourMapper.py     \
-file yourReducer.py   -reducer yourReducer.py    \
-input $yourInputFile    \
-output $yourOutputDirectoryOnHDFS

我还将该 jar 包含在我发出该调用的文件夹中。

它非常适合我的需求（并且它在减速器之后的行尾没有创建任何选项卡）。

更新：根据暗示这确实对其他人有帮助的评论，这是我的 CustomOutputFormat.java 文件的完整源代码：

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

import org.apache.hadoop.fs.FSDataOutputStream;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
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.TextOutputFormat;
import org.apache.hadoop.util.Progressable;
import org.apache.hadoop.util.ReflectionUtils;

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

    public RecordWriter<K, V> getRecordWriter(FileSystem ignored, JobConf job, String name,
        Progressable progress) throws IOException {
    boolean isCompressed = getCompressOutput(job);

    //Channging the default from '\t' to blank
    String keyValueSeparator = job.get("mapred.textoutputformat.separator", ""); // '\t'
    if (!isCompressed) {
        Path file = FileOutputFormat.getTaskOutputPath(job, name);
        FileSystem fs = file.getFileSystem(job);
        FSDataOutputStream fileOut = fs.create(file, progress);
        return new LineRecordWriter<K, V>(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 = FileOutputFormat.getTaskOutputPath(job, name + codec.getDefaultExtension());
        FileSystem fs = file.getFileSystem(job);
        FSDataOutputStream fileOut = fs.create(file, progress);
        return new LineRecordWriter<K, V>(new DataOutputStream(
            codec.createOutputStream(fileOut)), keyValueSeparator);
    }
    }
}

仅供参考：对于您的使用上下文，请务必检查这不会对映射器和化简器之间的 hadoop-streaming 托管交互（就分离键与值而言）产生不利影响。澄清：

• 根据我的测试 - 如果您的数据的每一行都有一个“选项卡”（每侧都有一些内容），您可以保留内置默认值：流式传输会将第一个选项卡之前的第一个内容解释为你的“键”，以及它后面那一行的所有内容作为你的“值”。因此，它不会看到“空值”，并且不会附加在您的化简器之后显示的选项卡。 （您将看到最终输出根据“键”的值进行排序，流将每行中的“键”解释为每个选项卡之前出现的值。）

• 相反，如果您的数据中没有选项卡，并且您没有使用上述技巧覆盖默认值，那么您将在运行完成后看到选项卡，上述覆盖将成为修复。