0

我 (title , text )使用 Java 从 Hadoop 中的 MapReduce 应用程序的输出中获得了几个有序对。

现在我想在这些有序对的文本字段上实现字数统计。

所以我的最终输出应该是这样的：

(title-a , word-a-1 , count-a-1 , word-a-2 , count-a-2 ....)

(title-b , word-b-1, count-b-1 , word-b-2 , count-b-2 ....)
.
.
.
.
(title-x , word-x-1, count-x-1 , word-x-2 , count-x-2 ....)

总而言之，我想在第一个 mapreduce 的输出记录上分别实现 wordcount。有人可以建议我一个好方法吗，或者我可以如何链接第二个 map reduce 工作来创建上述输出或更好地格式化它？

下面是代码，从github借来的，做了一些修改

package com.org;
import javax.xml.stream.XMLStreamConstants;//XMLInputFactory;
import java.io.*;
import java.io.IOException;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.fs.FSDataInputStream;
import org.apache.hadoop.fs.FSDataOutputStream;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.io.DataOutputBuffer;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.InputSplit;
import org.apache.hadoop.mapreduce.RecordReader;
import org.apache.hadoop.mapreduce.TaskAttemptContext;
import org.apache.hadoop.mapreduce.TaskAttemptID;
import org.apache.hadoop.mapreduce.lib.input.FileSplit;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;               
import org.apache.hadoop.mapreduce.lib.input.TextInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;
import javax.xml.stream.*;

public class XmlParser11
{

    public static class XmlInputFormat1 extends TextInputFormat {

    public static final String START_TAG_KEY = "xmlinput.start";
    public static final String END_TAG_KEY = "xmlinput.end";


    public RecordReader<LongWritable, Text> createRecordReader(
            InputSplit split, TaskAttemptContext context) {
        return new XmlRecordReader();
    }

    /**
     * XMLRecordReader class to read through a given xml document to output
     * xml blocks as records as specified by the start tag and end tag
     *
     */
    // @Override
    public static class XmlRecordReader extends
            RecordReader<LongWritable, Text> {
        private byte[] startTag;
        private byte[] endTag;
        private long start;
        private long end;
        private FSDataInputStream fsin;
        private DataOutputBuffer buffer = new DataOutputBuffer();

        private LongWritable key = new LongWritable();
        private Text value = new Text();
        @Override
        public void initialize(InputSplit split, TaskAttemptContext context)
                throws IOException, InterruptedException {
            Configuration conf = context.getConfiguration();
            startTag = conf.get(START_TAG_KEY).getBytes("utf-8");
            endTag = conf.get(END_TAG_KEY).getBytes("utf-8");
            FileSplit fileSplit = (FileSplit) split;

            // open the file and seek to the start of the split
            start = fileSplit.getStart();
            end = start + fileSplit.getLength();
            Path file = fileSplit.getPath();
            FileSystem fs = file.getFileSystem(conf);
            fsin = fs.open(fileSplit.getPath());
            fsin.seek(start);

        }
    @Override
    public boolean nextKeyValue() throws IOException,
                InterruptedException {
            if (fsin.getPos() < end) {
                if (readUntilMatch(startTag, false)) {
                    try {
                        buffer.write(startTag);
                        if (readUntilMatch(endTag, true)) {
                            key.set(fsin.getPos());
                            value.set(buffer.getData(), 0,
                                    buffer.getLength());
                            return true;
                        }
                    } finally {
                        buffer.reset();
                    }
                }
            }
            return false;
        }
    @Override
    public LongWritable getCurrentKey() throws IOException,
                InterruptedException {
            return key;
        }

    @Override
    public Text getCurrentValue() throws IOException,
                InterruptedException {
            return value;
        }
    @Override
    public void close() throws IOException {
            fsin.close();
        }
    @Override
        public float getProgress() throws IOException {
            return (fsin.getPos() - start) / (float) (end - start);
        }

        private boolean readUntilMatch(byte[] match, boolean withinBlock)
                throws IOException {
            int i = 0;
            while (true) {
                int b = fsin.read();
                // end of file:
                if (b == -1)
                    return false;
                // save to buffer:
                if (withinBlock)
                    buffer.write(b);
                // check if we're matching:
        if (b == match[i]) {
                    i++;
                    if (i >= match.length)
                        return true;
                } else
                    i = 0;
                // see if we've passed the stop point:
                if (!withinBlock && i == 0 && fsin.getPos() >= end)
                    return false;
            }
        }
    }
}


    public static class Map extends Mapper<LongWritable, Text,Text, Text> {
  @Override
  protected void map(LongWritable key, Text value,
                 Mapper.Context context)
  throws
  IOException, InterruptedException {
    String document = value.toString();
    System.out.println("'" + document + "'");
        try {
      XMLStreamReader reader =                      XMLInputFactory.newInstance().createXMLStreamReader(new     
           ByteArrayInputStream(document.getBytes()));
  String propertyName = "";
  String propertyValue = "";
  String currentElement = "";
  while (reader.hasNext()) {
    int code = reader.next();
    switch (code) {
      case XMLStreamConstants.START_ELEMENT: //START_ELEMENT:
        currentElement = reader.getLocalName();
        break;
      case XMLStreamConstants.CHARACTERS:  //CHARACTERS:
        if (currentElement.equalsIgnoreCase("title")) {
          propertyName += reader.getText();
          //System.out.println(propertyName);
        } else if (currentElement.equalsIgnoreCase("text")) {
     propertyValue += reader.getText();
          //System.out.println(propertyValue);
        }
        break;
    }
  }
  reader.close();
  context.write(new Text(propertyName.trim()), new Text(propertyValue.trim()));

}
    catch(Exception e){
            throw new IOException(e);

            }

  }
}
public static class Reduce
extends Reducer<Text, Text, Text, Text> {

  @Override
  protected void setup(
  Context context)
  throws IOException, InterruptedException {
    context.write(new Text("<Start>"), null);
  }

  @Override
  protected void cleanup(
  Context context)
  throws IOException, InterruptedException {
    context.write(new Text("</Start>"), null);
  }

  private Text outputKey = new Text();
  public void reduce(Text key, Iterable<Text> values,
                 Context context)
  throws IOException, InterruptedException {
for (Text value : values) {
      outputKey.set(constructPropertyXml(key, value));
      context.write(outputKey, null);
    }
 }

  public static String constructPropertyXml(Text name, Text value) {
    StringBuilder sb = new StringBuilder();
    sb.append("<property><name>").append(name)
    .append("</name><value>").append(value)
    .append("</value></property>");
    return sb.toString();
  }
}



    public static void main(String[] args) throws Exception
    {
            Configuration conf = new Configuration();

            conf.set("xmlinput.start", "<page>");
            conf.set("xmlinput.end", "</page>");
            Job job = new Job(conf);
            job.setJarByClass(XmlParser11.class);
            job.setOutputKeyClass(Text.class);
            job.setOutputValueClass(Text.class);

            job.setMapperClass(XmlParser11.Map.class);
            job.setReducerClass(XmlParser11.Reduce.class);

            job.setInputFormatClass(XmlInputFormat1.class);
            job.setOutputFormatClass(TextOutputFormat.class);

            FileInputFormat.addInputPath(job, new Path(args[0]));
            FileOutputFormat.setOutputPath(job, new Path(args[1]));

            job.waitForCompletion(true);
    }
}

我们在网上找到的 wordcount 代码会计算所有文件的字数并给出输出。我想分别对每个文本字段进行字数统计。上面的映射器用于从 XML 文档中提取标题和文本。有什么办法可以在同一个映射器中进行字数统计。如果我这样做，我的下一个疑问是如何将它与已经存在的键值对（标题，文本）一起传递给减速器。抱歉，我无法正确表达我的问题，但我想读者一定有所了解

4

2 回答 2

0

我建议你可以使用正则表达式

并执行映射和分组。在 hadoop 示例 jar 文件中提供使用此的 Grep 类，您可以使用正则表达式执行您的 hdfs 数据的映射。并将您的映射数据分组。

于 2013-09-16T13:25:55.133 回答
0

我不确定我是否理解正确。所以我有很多问题以及我的答案。

首先，编写此代码的人可能试图展示如何编写自定义 InputFormat以使用 MR 处理 xml 数据。我不知道它与你的问题有什么关系。

总而言之，我想在第一个 mapreduce 的输出记录上分别实现 wordcount。有人可以建议我一个好方法吗

读取第一个 MR 生成的输出文件并执行。

或者我如何可以链接第二个 map reduce 作业来创建上述输出或更好地格式化它？

您绝对可以通过编写多个驱动程序方法以这种方式将作业链接在一起，每个作业一个。有关更多详细信息，请参阅this ，并以this为例。

我想分别对每个文本字段进行字数统计。

分开是什么意思？在传统的 wordcount 程序中，每个单词的计数是独立于其他单词计算的。

有什么办法可以在同一个映射器中进行字数统计。

我希望你已经正确理解了 wordcount 程序。在传统的 wordcount 程序中，您读取输入文件，一次一行，将行分割成单词，然后以1 作为值发出每个单词作为键。所有这些都发生在 Mapper 内部，它本质上是同一个 Mapper。然后每个单词的总计数在您工作的 Reducer 部分中确定。如果您希望从映射器本身发出带有总计数的单词，您必须读取映射器本身中的整个文件并进行计数。为此，您需要将 InputFormat 中的 isSplittable 设置为false ，以便将您的输入文件作为一个整体读取并仅转到一个 Mapper。

当您从 Mapper 发出某些内容并且它不是仅 Map 的作业时，您的 Mapper 的输出会自动转到 Reducer。您还需要些别的吗？

于 2013-08-01T20:48:49.610 回答