Java实现Hadoop MapReduce计算各产品编码平均订单需求的问题求助

核心报错/输出异常原因 #

• Reducer核心逻辑未被执行（最关键）
  类继承泛型参数完全错位：FrequencyReducer定义为Reducer<Text, IntWritable, IntWritable, FloatWritable>，代表输出key是IntWritable类型，但Mapper输出的key是Text类型的商品编码，Driver里也明确配置最终输出key为Text类型，类型完全不匹配。
  导入类错误：误导入javax.xml.soap.Text而非Hadoop原生的org.apache.hadoop.io.Text，存在编译兼容隐患。
  方法签名错误：写的reduce方法第一个参数是IntWritable productcode，和框架要求的“reduce方法第一个参数与Mapper输出key类型一致（Text）”的规则不符，属于方法重载而非方法重写，Hadoop框架根本不会调用这段写了求平均逻辑的代码，最终执行默认的Reducer透传逻辑，完全达不到聚合效果。
• Mapper的有效行判断逻辑有漏洞
  用if(line.contains("Product"))做过滤，会把CSV第一行表头（内容本身包含"Product_Code"）当成有效数据解析，既会把"Product_Code"当商品编码写入输出，又会尝试把字符串"Order_Demand"转成数字，直接抛出数字格式异常。
• CSV字段解析未做格式兼容
  示例数据里的Order_Demand字段末尾带多余空格（比如500 、160000 ），直接调用数字解析方法会因为空格报错；同时把订单值强制转int存储，带小数的订单值会被截断丢精度。

修正后可正常运行的代码 #

FrequencyMapper.java #

package ma.test.a02;

import java.io.IOException;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;

public class FrequencyMapper extends Mapper<LongWritable, Text, Text, IntWritable> {
    private static final String HEADER_TAG = "Product_Code";

    @Override
    protected void map(LongWritable offset, Text lineText, Context context)
            throws IOException, InterruptedException {
        String line = lineText.toString().trim();
        if (line.isEmpty()) return;
        String[] fields = line.split(",");
        // 跳过表头行
        if (HEADER_TAG.equals(fields[0])) return;
        String productCode = fields[0].trim();
        // 去掉字段前后空格再解析数字
        int orderDemand = Integer.parseInt(fields[4].trim());
        context.write(new Text(productCode), new IntWritable(orderDemand));
    }
}

FrequencyReducer.java #

package ma.test.a02;

import java.io.IOException;
import org.apache.hadoop.io.FloatWritable;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;

// 泛型顺序：输入key类型，输入value类型，输出key类型，输出value类型
public class FrequencyReducer extends Reducer<Text, IntWritable, Text, FloatWritable> {
    @Override
    protected void reduce(Text productCode, Iterable<IntWritable> orderDemands, Context context)
            throws IOException, InterruptedException {
        long sum = 0;
        int count = 0;
        for (IntWritable demand : orderDemands) {
            sum += demand.get();
            count++;
        }
        float average = (float) sum / count;
        context.write(productCode, new FloatWritable(average));
    }
}

Driver类无需改动 #

原有Driver代码的类型配置、作业设置完全符合规范，和修正后的Mapper、Reducer可以完全适配。

补充优化提示：如果业务中订单需求数值超过int存储范围（大于2147483647），可以把Map输出的value类型替换为LongWritable，避免整数溢出。

内容的提问来源于stack exchange，提问作者ems58fly