我正在为map和reduce编写代码，以便以后在hadoop上运行。我必须得到3列：
人员id，
他演过多少部电影，
他导演的电影数量
数据文件如下所示：
tconst（字符串）
标题的字母数字唯一标识符
排序（整数）
唯一标识给定标题ID行的数字
nconst（字符串）
姓名/人员的字母数字唯一标识符
类别（字符串）
此人所从事的工作类别
作业（字符串）
具体的职务（如果适用），否则为“\n”
字符（字符串）
播放的角色的名称（如果适用），否则为'\n'

我在前面的练习中对wordcount.java的代码做了一些修改。

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

//        private Text year = new Text();
//        private IntWritable size = new IntWritable();
        private IntWritable size = new IntWritable();
        private Text category = new Text();

        public void map(LongWritable offset, Text lineText, Context context)

                throws IOException, InterruptedException {
            try {
                if (offset.get() == 0)
                    return;
                else {
                    String line = lineText.toString();
                    int i = 0;
                    for (String word : line
                            .split(",(?=([^\"]*\"[^\"]*\")*[^\"]*$)")) {
                        if (i == 4) {
                            year.set(word.substring(word.lastIndexOf('/') + 1,
                                    word.lastIndexOf('/') + 5));
                        }
                        if (i == 5) {
                            size.set(Integer.parseInt(word));
                        }
                        i++;
                    }
                    context.write(year, size);
                }
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
    }

    public static class AvgSizeStationReducer extends
            Reducer<Text, IntWritable, Text, DoubleWritable> {

        private DoubleWritable result = new DoubleWritable();
        Float average;
        Float count;
        int sum;

        @Override
        public void reduce(Text key, Iterable<IntWritable> values,
                           Context context) throws IOException, InterruptedException {
            average = 0f;
            count = 0f;
            sum = 0;

            Text sumText = new Text("average size of station for " + key
                    + " year is: ");

            for (IntWritable val : values) {
                sum += val.get();
                count += 1;
            }
            average = sum / count;
            result.set(average);
            context.write(sumText, result);
        }
     }

我如何做我的Map和缩小？