如何在 Hadoop 中创建自定义输出格式

Question

我正在尝试创建单词计数 hadoop 程序的变体，在该程序中它读取目录中的多个文件并输出每个单词的频率。问题是，我希望它输出一个单词，后跟文件名的来源以及该文件的频率。例如：

word1
( file1, 10)
( file2, 3)
( file3, 20)

所以对于 word1（说“和”这个词）。它发现 10 次是 file1，3 次是 file2，等等。现在它只输出一个键值对

 StringTokenizer itr = new StringTokenizer(chapter);
  while (itr.hasMoreTokens()) {
    word.set(itr.nextToken());

    context.write(word, one);

我可以通过

获取文件名

String fileName = ((FileSplit) context.getInputSplit()).getPath().getName();

但我不明白如何按照我想要的方式进行格式化。我一直在研究 OutputCollector，但我不确定如何准确地使用它。

编辑：这是我的映射器和recuder

public static class TokenizerMapper
   extends Mapper<Object, Text, Text, Text>{ 

private Text word = new Text();

public void map(Object key, Text value, Context context
                ) throws IOException, InterruptedException {

  //Take out all non letters and make all lowercase
  String chapter = value.toString();
  chapter = chapter.toLowerCase();
  chapter = chapter.replaceAll("[^a-z]"," ");

  //This is the file name
  String fileName = ((FileSplit) context.getInputSplit()).getPath().getName();

  StringTokenizer itr = new StringTokenizer(chapter);
  while (itr.hasMoreTokens()) {
    word.set(itr.nextToken());

   context.write(word, new Text(fileName)); //
  }
}
  }


  public static class IntSumReducer
       extends Reducer<Text,Text,Text,Text> { second


   public void reduce(Text key, Iterable<Text> values, Context context)
         throws IOException, InterruptedException {

  Map<String, Integer> files = new HashMap<String, Integer>();

 for (Text val : values) {
    if (files.containsKey(val.toString())) {
        files.put(val.toString(), files.get(val.toString())+1);
    } else {
        files.put(val.toString(), 1); 
    }
}

String outputString="";

for (String file : files.keySet()) { 
    outputString = outputString + "\n<" + file + ", " + files.get(file) + ">"; //files.get(file)
}

context.write(key, new Text(outputString));
}

  }

这是为单词“a”输出的，例如：

a   
(
(chap02, 53), 1)
(
(chap18, 50), 1)

我不确定为什么它使键值对成为每个条目的值 1 的键。

Answer 1

我认为您根本不需要自定义输出格式。只要您将文件名传递给减速器，您应该能够通过修改您在 TextOutputFormat 类型操作中使用的 String 来完成此操作。说明如下。

在映射器中获取文件名，并将其附加到 textInputFormat 如下

String fileName = ((FileSplit) context.getInputSplit()).getPath().getName();
context.write(key,new Text(fileName));

然后在 reducer 中执行如下操作：

public void reduce(Text key, Iterable<Text> values, Context context)
        throws IOException, InterruptedException {
    Map<String, Integer> files = new HashMap<String, Integer>();
    for (Text val : values) {
        if (files.containsKey(val.toString())) {
            files.put(val.toString(), files.get(val.toString()) + 1);
        } else {
            files.put(val.toString(), 1);
        }
    }

    String outputString = key.toString();

    for (String file : files.keySet()) {
        outputString += "\n( " + file + ", " + files.get(file) + ")";
    }

    context.write(key, new Text(outputString));
}

这个reducer 将"\n" 附加到每一行的开头，以强制显示格式完全符合您的要求。

这似乎比编写自己的输出格式要简单得多。