Hadoop для записи readFields EOFException

Question

Я реализую свою собственную вторичную сортировку Writable for Hadoop, но при запуске задания Hadoop продолжает выдавать исключение EOFException в моем readFields метод, и я не знаю, что с ним не так.

Трассировка стека ошибок:

java.lang.Exception: java.lang.RuntimeException: java.io.EOFException
    at org.apache.hadoop.mapred.LocalJobRunner$Job.runTasks(LocalJobRunner.java:492)
    at org.apache.hadoop.mapred.LocalJobRunner$Job.run(LocalJobRunner.java:559)
Caused by: java.lang.RuntimeException: java.io.EOFException
    at org.apache.hadoop.io.WritableComparator.compare(WritableComparator.java:165)
    at org.apache.hadoop.mapreduce.task.ReduceContextImpl.nextKeyValue(ReduceContextImpl.java:158)
    at org.apache.hadoop.mapreduce.task.ReduceContextImpl.nextKey(ReduceContextImpl.java:121)
    at org.apache.hadoop.mapreduce.lib.reduce.WrappedReducer$Context.nextKey(WrappedReducer.java:302)
    at org.apache.hadoop.mapreduce.Reducer.run(Reducer.java:170)
    at org.apache.hadoop.mapred.ReduceTask.runNewReducer(ReduceTask.java:628)
    at org.apache.hadoop.mapred.ReduceTask.run(ReduceTask.java:390)
    at org.apache.hadoop.mapred.LocalJobRunner$Job$ReduceTaskRunnable.run(LocalJobRunner.java:347)
    at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511)
    at java.util.concurrent.FutureTask.run(FutureTask.java:266)
    at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142)
    at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617)
    at java.lang.Thread.run(Thread.java:745)
Caused by: java.io.EOFException
    at java.io.DataInputStream.readInt(DataInputStream.java:392)
    at org.apache.hadoop.io.IntWritable.readFields(IntWritable.java:47)
    at writable.WikiWritable.readFields(WikiWritable.java:39)
    at org.apache.hadoop.io.WritableComparator.compare(WritableComparator.java:158)
    ... 12 more

Мой код:

package writable;

import org.apache.hadoop.io.*;

import java.io.DataInput;
import java.io.DataOutput;
import java.io.IOException;

public class WikiWritable implements WritableComparable<WikiWritable> {
  private IntWritable docId;
  private IntWritable position;

  public WikiWritable() {
    this.docId = new IntWritable();
    this.position = new IntWritable();
  }

  public void set(String docId, int position) {
    this.docId = new IntWritable(Integer.valueOf(docId));
    this.position = new IntWritable(position);
  }

  @Override
  public int compareTo(WikiWritable o) {
    int result = this.docId.compareTo(o.docId);
    result = result == 0 ? this.position.compareTo(o.position) : result;
    return result;
  }

  @Override
  public void write(DataOutput dataOutput) throws IOException {
    docId.write(dataOutput);
    position.write(dataOutput); // error here
  }

  @Override
  public void readFields(DataInput dataInput) throws IOException {
    docId.readFields(dataInput);
    position.readFields(dataInput);
  }

  public IntWritable getDocId() {
    return docId;
  }

  public int getPosition() {
    return Integer.valueOf(position.toString());
  }
}

// Driver
public class Driver {
  public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException {
    Path wiki = new Path(args[0]);
    Path out = new Path(args[1]);

    Configuration conf = new Configuration();
    Job job = Job.getInstance(conf, "myjob");

    TextInputFormat.addInputPath(job, wiki);
    TextOutputFormat.setOutputPath(job, out);

    job.setMapOutputKeyClass(Text.class);
    job.setMapOutputValueClass(WikiWritable.class);

    job.setJarByClass(Driver.class);
    job.setMapperClass(WordMapper.class);
    job.setOutputKeyClass(Text.class);
    job.setOutputValueClass(Text.class);
    job.setReducerClass(WordReducer.class);
    job.setPartitionerClass(WikiPartitioner.class);
    job.setGroupingComparatorClass(WikiComparator.class);

    job.waitForCompletion(true);
  }
}

// Mapper.map
protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
    String[] words = value.toString().split(",");
    String id = words[0];
    String[] contents = words[3].toLowerCase().replaceAll("[^a-z]+", " ").split("\\s+");

    for (int i = 0; i < contents.length; i++) {
      String word = contents[i].trim();
      word = stem(word);
      WikiWritable output = new WikiWritable();
      output.set(id, i);
      context.write(new Text(contents[i]), output);
    }
  }

// Comparator
public class WikiComparator extends WritableComparator {
  public WikiComparator() {
    super(WikiWritable.class, true);
  }

  @Override
  public int compare(WritableComparable wc1, WritableComparable wc2) {
    WikiWritable w1 = (WikiWritable) wc1;
    WikiWritable w2 = (WikiWritable) wc2;
    return w1.compareTo(w2);
  }
}

// Partitioner
public class WikiPartitioner extends Partitioner<WikiWritable, Text> {
  @Override
  public int getPartition(WikiWritable wikiWritable, Text text, int i) {
    return Math.abs(wikiWritable.getDocId().hashCode() % i);
  }
}

// Reducer
public class WordReducer extends Reducer<Text, WikiWritable, Text, Text> {
  @Override
  protected void reduce(Text key, Iterable<WikiWritable> values, Context ctx) throws IOException, InterruptedException {
    Map<String, StringBuilder> map = new HashMap<>();
    for (WikiWritable w : values) {
      String id = String.valueOf(w.getDocId());
      if (map.containsKey(id)) {
        map.get(id).append(w.getPosition()).append(".");
      } else {
        map.put(id, new StringBuilder());
        map.get(id).append(".").append(w.getPosition()).append(".");
      }
    }

    StringBuilder builder = new StringBuilder();
    map.keySet().forEach((k) -> {
      map.get(k).deleteCharAt(map.get(k).length() - 1);
      builder.append(k).append(map.get(k)).append(";");
    });

    ctx.write(key, new Text(builder.toString()));
  }
}

При строительстве нового WikiWritableмаппер сначала звонит new WikiWritable() а потом звонит set(...),

Я пытался изменить docId а также position String и Integer и использовать dataOutput.read() (Я забыл точное имя метода, но это что-то похожее) и до сих пор не работает.

Answer 1

TLDR: Вам просто нужно удалить свой WikiComparator полностью, а не звонить job.setGroupingComparatorClass совсем.

Объяснение: Компаратор группы предназначен для сравнения выходных ключей карты, а не выходных значений карты. Ваши выходные ключи карты Text объекты и значения WikiWritable объекты. Это означает, что байты, которые передаются вашему компаратору для десериализации, представляют собой сериализованные Text объекты. Тем не менее WikiComparator использует отражение для создания WikiWritable объекты (как указано в его конструкторе), а затем пытается десериализовать Text объекты, использующие WikiWritable.readFields метод. Это, очевидно, приводит к неправильному чтению и, следовательно, к исключению, которое вы видите.

Тем не менее, я считаю, что вам не нужен компаратор вообще, так как по умолчанию WritableComparator делает именно то, что делает ваш: вызывает compareTo метод для пары объектов, которые передаются ему.

РЕДАКТИРОВАТЬ: compareTo вызываемый метод сравнивает ваши ключи, а не значения, поэтому он сравнивает Text объекты. Если вы хотите сравнить и отсортировать WikiWritables вы должны рассмотреть возможность добавления их в составной ключ. Существует множество обучающих программ по составным ключам и вторичной сортировке.