java编写话费统计的MR程序，计算每个电话号码的总流量，包括上行流量，下行流量。

好的，这个问题需要用到 Hadoop 的 MapReduce 编程模型，以下是一个简单的 Java 代码实现：

首先，在 Map 阶段，我们需要将每个电话号码作为 key，上下行流量作为 value，输出到 Reduce 阶段进行统计。代码如下：

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

  private Text phoneNumber = new Text();
  private IntWritable flow = new IntWritable();

  public void map(Object key, Text value, Context context) throws IOException, InterruptedException {
    String[] line = value.toString().split(",");
    phoneNumber.set(line[0]);
    flow.set(Integer.parseInt(line[1]) + Integer.parseInt(line[2]));
    context.write(phoneNumber, flow);
  }
}

接着，在 Reduce 阶段，我们将每个电话号码的流量值进行累加，最终输出每个电话号码的总流量。代码如下：

public static class SumReducer extends Reducer<Text, IntWritable, Text, IntWritable> {

  private IntWritable result = new IntWritable();

  public void reduce(Text key, Iterable<IntWritable> values, Context context) throws IOException, InterruptedException {
    int sum = 0;
    for (IntWritable val : values) {
      sum += val.get();
    }
    result.set(sum);
    context.write(key, result);
  }
}

最后，在 main 函数中配置 MapReduce 作业的一些参数并提交任务。代码如下：

public static void main(String[] args) throws Exception {
  Configuration conf = new Configuration();
  Job job = Job.getInstance(conf, "flow count");
  job.setJarByClass(FlowCount.class);
  job.setMapperClass(TokenizerMapper.class);
  job.setCombinerClass(SumReducer.class);
  job.setReducerClass(SumReducer.class);
  job.setOutputKeyClass(Text.class);
  job.setOutputValueClass(IntWritable.class);
  FileInputFormat.addInputPath(job, new Path(args[0]));
  FileOutputFormat.setOutputPath(job, new Path(args[1]));
  System.exit(job.waitForCompletion(true) ? 0 : 1);
}

其中，args[0] 和 args[1] 分别表示输入文件路径和输出文件路径。运行这个程序，就可以得到每个电话号码的总流量统计结果了。