Как форматировать вывод, записываемый Mapreduce в Hadoop

Question

Я пытаюсь изменить содержимое файла на каждое слово. У меня есть программа работает нормально, но на выходе я получаю что-то вроде этого

1 dwp 
2 seviG 
3 eht 
4 tnerruc 
5 gnikdrow 
6 yrotcerid 
7 ridkm 
8 desU 
9 ot 
10 etaerc 

Я хочу выход быть что-то вроде этого

dwp seviG eht tnerruc gnikdrow yrotcerid ridkm desU 
ot etaerc 

Код я работаю с

import java.io.IOException; 
    import java.util.*; 

    import org.apache.hadoop.fs.Path; 
    import org.apache.hadoop.conf.*; 
    import org.apache.hadoop.io.*; 
    import org.apache.hadoop.mapred.*; 
    import org.apache.hadoop.util.*; 

    public class Reproduce { 

    public static int temp =0; 
    public static class ReproduceMap extends MapReduceBase implements Mapper<LongWritable, Text, IntWritable, Text>{ 
     private Text word = new Text(); 
     @Override 
     public void map(LongWritable arg0, Text value, 
       OutputCollector<IntWritable, Text> output, Reporter reporter) 
       throws IOException { 
      String line = value.toString().concat("\n"); 
      StringTokenizer tokenizer = new StringTokenizer(line); 
      while (tokenizer.hasMoreTokens()) { 
       word.set(new StringBuffer(tokenizer.nextToken()).reverse().toString()); 
       temp++; 
       output.collect(new IntWritable(temp),word); 
       } 

     } 

    } 

    public static class ReproduceReduce extends MapReduceBase implements Reducer<IntWritable, Text, IntWritable, Text>{ 

     @Override 
     public void reduce(IntWritable arg0, Iterator<Text> arg1, 
       OutputCollector<IntWritable, Text> arg2, Reporter arg3) 
       throws IOException { 
      String word = arg1.next().toString(); 
      Text word1 = new Text(); 
      word1.set(word); 
      arg2.collect(arg0, word1); 

     } 

    } 

    public static void main(String[] args) throws Exception { 
    JobConf conf = new JobConf(WordCount.class); 
    conf.setJobName("wordcount"); 

    conf.setOutputKeyClass(IntWritable.class); 
    conf.setOutputValueClass(Text.class); 

    conf.setMapperClass(ReproduceMap.class); 
    conf.setReducerClass(ReproduceReduce.class); 

    conf.setInputFormat(TextInputFormat.class); 
    conf.setOutputFormat(TextOutputFormat.class); 

    FileInputFormat.setInputPaths(conf, new Path(args[0])); 
    FileOutputFormat.setOutputPath(conf, new Path(args[1])); 

    JobClient.runJob(conf); 

    } 
} 

Как изменить свой вывод вместо того, чтобы писать другую программу java для этого

Thank s заранее

Answer 1

Вот простой код демонстрирует использование пользовательского FileoutputFormat

public class MyTextOutputFormat extends FileOutputFormat<Text, List<IntWritable>> { 
     @Override 
     public org.apache.hadoop.mapreduce.RecordWriter<Text, List<Intwritable>> getRecordWriter(TaskAttemptContext arg0) throws IOException, InterruptedException { 
     //get the current path 
     Path path = FileOutputFormat.getOutputPath(arg0); 
     //create the full path with the output directory plus our filename 
     Path fullPath = new Path(path, "result.txt"); 
    //create the file in the file system 
    FileSystem fs = path.getFileSystem(arg0.getConfiguration()); 
    FSDataOutputStream fileOut = fs.create(fullPath, arg0); 

    //create our record writer with the new file 
    return new MyCustomRecordWriter(fileOut); 
    } 
} 

public class MyCustomRecordWriter extends RecordWriter<Text, List<IntWritable>> { 
    private DataOutputStream out; 

    public MyCustomRecordWriter(DataOutputStream stream) { 
     out = stream; 
     try { 
      out.writeBytes("results:\r\n"); 
     } 
     catch (Exception ex) { 
     } 
    } 

    @Override 
    public void close(TaskAttemptContext arg0) throws IOException, InterruptedException { 
     //close our file 
     out.close(); 
    } 

    @Override 
    public void write(Text arg0, List arg1) throws IOException, InterruptedException { 
     //write out our key 
     out.writeBytes(arg0.toString() + ": "); 
     //loop through all values associated with our key and write them with commas between 
     for (int i=0; i<arg1.size(); i++) { 
      if (i>0) 
       out.writeBytes(","); 
      out.writeBytes(String.valueOf(arg1.get(i))); 
     } 
     out.writeBytes("\r\n"); 
    } 
} 

Наконец, мы должны сказать, свою работу о нашем формате Ouput и пути перед его запуском.

job.setOutputKeyClass(Text.class); 
job.setOutputValueClass(ArrayList.class); 
job.setOutputFormatClass(MyTextOutputFormat.class); 
FileOutputFormat.setOutputPath(job, new Path("/home/hadoop/out")); 

Answer 2

вы можете использовать NullWritable в качестве выходного значения. NullWritable - это просто заполнитель. Поскольку вы не хотите, чтобы число отображалось как часть вашего вывода. Я дал модифицированный класс редуктора. Примечание: - необходимо добавить оператор импорта для NullWritable

public static class ReproduceReduce extends MapReduceBase implements Reducer<IntWritable, Text, Text, NullWritable>{ 

      @Override 
      public void reduce(IntWritable arg0, Iterator<Text> arg1, 
        OutputCollector<Text, NullWritable> arg2, Reporter arg3) 
        throws IOException { 
       String word = arg1.next().toString(); 
       Text word1 = new Text(); 
       word1.set(word); 
       arg2.collect(word1, new NullWritable()); 

      } 

     } 

и изменить класс драйвера или основной метод

conf.setOutputKeyClass(Text.class); 
conf.setOutputValueClass(NullWritable.class); 

Answer 3

В ключевой темп Mapper увеличивается для каждого значения слова, поэтому каждое слово обрабатывается как отдельная пара ключ-значение.

Ниже следует решить проблему 1) В Mapper просто удалите temp ++, так что все обратные слова будут иметь ключ как 0 (temp = 0).

2) Редуктор получает ключ 0 и список обратных строк. В редукторе установите ключ на NullWritable и запишите выходной сигнал.

Answer 4

Мы можем настроить вывод, написав пользовательские fileoutputformat класс

Answer 5

То, что вы можете попробовать это сделать один постоянный ключ (или просто nullwritable) и передать его в качестве ключа и вашей полной линии в качестве значения (можно отменить это в классе сопоставления или вы также можете отменить его и в классе редуктора). поэтому ваш редуктор получит постоянный ключ (или держатель места, если вы использовали nullwritable в качестве ключа) и полную строку. Теперь вы можете просто перевернуть строку и записать ее в выходной файл. Не используя tmp в качестве ключа, вы избегаете записи нежелательных чисел в выходной файл.