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

好的，这是一个基于Hadoop平台下的MapReduce程序。您需要准备好Hadoop的环境和Java开发环境。

首先，我们需要定义Mapper类和Reducer类。Mapper类的作用是将输入的数据分割成小块，以便于后续处理，Reducer类的作用是对Mapper处理后的结果进行汇总。

Mapper类的代码如下所示：

public class FlowMapper extends Mapper<LongWritable, Text, Text, LongWritable> {
    private Text phoneNum = new Text();
    private LongWritable flow = new LongWritable();

    public void map(LongWritable key, Text value, Context context)
            throws IOException, InterruptedException {
        String line = value.toString();
        String[] items = line.split(",");
        String num = items[0];
        long upFlow = Long.parseLong(items[1]);
        long downFlow = Long.parseLong(items[2]);
        flow.set(upFlow + downFlow);
        phoneNum.set(num);
        context.write(phoneNum, flow);
    }
}

Mapper类中，我们定义了一个Text类型的变量phoneNum和一个LongWritable类型的变量flow，分别表示电话号码和总流量。在map()函数中，我们首先将输入的一行数据转换成字符串类型，并将其按照逗号分割成若干项。然后，我们将第一项作为电话号码，将第二项和第三项相加得到总流量，并将其赋值给flow变量。最后，我们将电话号码和总流量写入到上下文中。

Reducer类的代码如下所示：

public class FlowReducer extends Reducer<Text, LongWritable, Text, LongWritable> {
    private LongWritable sumFlow = new LongWritable();

    public void reduce(Text key, Iterable<LongWritable> values, Context context)
            throws IOException, InterruptedException {
        long sum = 0;
        for (LongWritable value : values) {
            sum += value.get();
        }
        sumFlow.set(sum);
        context.write(key, sumFlow);
    }
}

Reducer类中，我们定义了一个LongWritable类型的变量sumFlow，表示总流量。在reduce()函数中，我们首先遍历values中的所有元素，将它们的值相加得到总流量。然后，我们将电话号码和总流量写入到上下文中。

最后，我们需要编写Driver类，用于启动MapReduce程序。Driver类的代码如下所示：

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

Driver类中，我们首先定义了一个Configuration对象和一个Job对象。然后，我们设置了MapReduce程序的相关参数，包括Mapper、Reducer、输入格式、输出格式等。最后，我们调用job.waitForCompletion()方法启动MapReduce程序，并等待程序执行完成。

以上就是用Java编写话费统计的MapReduce程序的全部内容。您可以将代码保存为FlowCount.java文件，并使用hadoop com.sun.tools.javac.Main FlowCount.java命令进行编译。然后，使用jar命令将编译后的class文件打包成jar包，并使用hadoop jar命令启动MapReduce程序。