mapのキーで行を選択的に選択する方法reduce

Question

私は以下のcsvファイルを5列あり、多くの行を持っています。しかし、私は最初の6行だけを表示しています。

Date,Food,Vitamin,Protein,NumStudents 
01/01/17, Pasta, A, Yes, 560 
01/01/17, Pizza, A, Yes, 730 
01/01/17, Burrito, C, Yes, 240 
02/01/17, Pizza, A, Yes, 340 
02/01/17, Pasta, B, Yes, 450 
02/01/17, Beef, B, Yes, 450 

ここで、ピザとパスタだけを持っている特定の日に、NumStudentsの合計を求めたいと思います。

本質的には、01/01/17の場合、ピッツァとパスタにはNumStudentsを加算するだけですが、ブリットは加算しないでください。私は食品のすべての3つのタイプのためにNumStudentsを合計することができるよしかし、選択的に排除する方法がわからない私のコードでは

01/01/17 1530 
02/01/17 1240 

を取得しています出力

01/01/17 1290 
02/01/17 790 

出力を期待

マッパーの私のコンポジットキーからの何種類かの食べ物。どのように私はそれについて行く必要がありますどのようなアイデア？以下

私のコードは

public class GroupMR { 

    public static class GroupMapper extends Mapper<LongWritable, Text, DateYear, IntWritable> { 




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

      String line = value.toString(); 
      String[] keyvalue = line.split(","); 
      monyeartext.set(new Text(keyvalue[0])); 

      //populat.set(Integer.parseInt(keyvalue[5])); 
      termText.set(keyvalue[1]); 
      try { 
       numpass.set(Integer.parseInt(keyvalue[4])); 
      }catch (NumberFormatException e){ 
       System.out.println("not a number"); 
      } 
      DateYear monyear = new DateYear(monyeartext, termText); 
      context.write(monyear, numpass); 

     } 
    } 

Answer 1

のString []キー値= line.split（ ""）です。

Please add a filter after this line 

if(!(keyvalue[2].equals("Pasta") ||keyvalue[2].equals("Pizza"))){ 
    // If the food item is not pizza or pasta then return 
    return; 
} 

Answer 2

私はコードを減らしました。私は月と日付だけを選び、年に基づいてソートしました。

public class GroupMR { 

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

    Text numpass = null; 
    Text monyeartext = null; 

    public void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException { 
     String line = value.toString(); 
     String[] keyvalue = line.split(","); 
     String[] monyeartext1 = keyvalue[0].split("/"); 
     monyeartext = new Text(monyeartext1[2] + "/" +monyeartext1[0]); 
     numpass = new Text(keyvalue[1] + "-" + keyvalue[4]); 
     context.write(monyeartext, numpass); 

    } 
} 

public static class GroupReducer extends Reducer<Text, Text, Text, IntWritable> { 
    public void reduce(Text key, Iterable<Text> values, Context context) 
      throws IOException, InterruptedException { 
     boolean doesExist = false; 
     int sum = 0;   
     for (Text val : values) { 
      String[] val2 = val.toString().split("-"); 
      if (val2[0].equals("Pizza") || val2[0].equals("Pasta")){ 
       sum += Integer.parseInt(val2[1]); 
      } 
     } 
     context.write(key, new IntWritable(sum)); 
    } 
} 

public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException { 
    Configuration conf = new Configuration(); 
    String[] otherArgs = new GenericOptionsParser(conf, args).getRemainingArgs(); 
    if (otherArgs.length < 2) { 
     System.err.println("Usage: wordcount <in> [<in>...] <out>"); 
     System.exit(2); 
    } 
    Job job = Job.getInstance(conf, "GroupMR"); 
    job.setJarByClass(GroupMR.class); 
    job.setMapperClass(GroupMapper.class); 

    job.setReducerClass(GroupReducer.class); 
    job.setOutputKeyClass(Text.class); 
    job.setOutputValueClass(Text.class); 

    //for (int i = 0; i < otherArgs.length - 1; ++i) { 
     FileInputFormat.addInputPath(job, new Path(otherArgs[0])); 
    // } 

    FileOutputFormat.setOutputPath(job, new Path(otherArgs[1])); 
    System.exit(job.waitForCompletion(true) ? 0 : 1); 
} 

}