hadoop - 有点挑战性但很有趣的话题

Question

我们必须编写一个 InputFormat 和 RecordReader
来读取文本字符串
由“$”字符而不是换行符分隔。
现在假设我们有一个数据集像$ <some data> $ <some data>

我们想使用第一个 $ 作为分隔符，我们必须免除中间的“美元”符号使其成为分隔符。

您能否建议实现此目标的必要功能或必要步骤？
提前致谢

Answer 1

输入文件

$Aniruddha Sinha$23$Hadoop$mapreduce$Kishore$30$Hadoop$mapreduce

输出文件

Aniruddha Sinha,23,Hadoop,mapreduce
Kishore,30,Hadoop,mapreduce

代码

import java.io.IOException;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapred.SkipBadRecords;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

public class WordCount {

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

    private final static IntWritable one = new IntWritable(1);
    private Text word = new Text();
    private int count = 0;
    private StringBuilder sb = new StringBuilder();
    public void map(Object key, Text value, Context context
                    ) throws IOException, InterruptedException {
      if(count < 4){
          sb.append(value.toString());
          sb.append(",");
          count++;
          if(count == 4){
              System.out.println(sb.substring(0, sb.length()-1));
              count = 0;
              word.set(sb.substring(0, sb.length()-1));
              context.write(word, one);
              sb.delete(0, sb.length());
          }
      }
    }
  }


  public static void main(String[] args) throws Exception {
    Configuration conf = new Configuration();
    conf.set("textinputformat.record.delimiter", "$");
    Job job = Job.getInstance(conf, "word count");
    job.setJarByClass(WordCount.class);
    job.setMapperClass(TokenizerMapper.class);
    job.setOutputKeyClass(Text.class);
    job.setOutputValueClass(IntWritable.class);

    SkipBadRecords.setMapperMaxSkipRecords(conf, 10);
    FileInputFormat.setInputPaths(job, new Path("/home/kishore/input"));
    FileOutputFormat.setOutputPath(job, new Path("/home/kishore/output"));
    System.exit(job.waitForCompletion(true) ? 0 : 1);
  }
}