-sigma48 12mar09.9010.9010.3；k、 Tafqealdaagdklvvdfsatwc[160.14]gpc[160.14]k.m；p08263.3；1.062 sigma48\ U 12mar09.9063.9063.3；k、 kdpeglflqdnivaefsvdetgqmsatak.g；p08263.3；1.062
输入文件2数据
第1栏；第2栏；第3栏；第4栏；

-sigma48 12mar09.9188.9188.2；r、 yklslefpsgypynaptvk.f公司；p08263.3；1.062 sigma48\ U 12mar09.9314.9314.2；r、 yklslefpsgypynaptvk.fp08263.3；1.062 sigma48\ U 12mar09.9010.9010.3；k、 Tafqealdaagdklvvdfsatwc[160.14]gpc[160.14]k.m；p08263.3；1.062
我的要求：
读取input file1 data.csv中的所有行取column1并读取input file2 data.csv中的所有行，然后将第一个文件中的column1与第二个文件中的column1进行比较。
找到匹配项后，比较上述两个文件中的所有其他列，并将匹配的数据写入hdfs，并应返回这两个输入文件中匹配的百分比。
代码如下。。

/* First Mapper */
public void map(LongWritable key,Text value,Context context) 
            throws IOException, InterruptedException{

        String line = value.toString();
        String[] words = line.split(";");
        String name = words[1];
        String other = words[2];
        context.write(new Text(name), new Text(line));

    }
}

  /* Second Mapper */

public static class InputMapper2 extends Mapper<LongWritable,Text,Text,Text>{

    public void map(LongWritable key, Text value, Context context)
              throws IOException, InterruptedException
              {
        String line = value.toString();

        String[] words = line.split(";");
        String name = words[1];
        String other = words[2];
        System.out.println(key);
        context.write(new Text(name), new Text(line));

              }

              }

  /* Reducer for both of the mappers */
  /*incomplete and have to compare the two csv files here */

public static class CounterReducer extends Reducer
 {
  String line=null;

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

    Iterator<Text> val = values.iterator();

   for(Text value:values)
   {
    line = value.toString();
   }

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