在Hadoop中实现BigDecimal相加的最优方案是什么？

既然Apache Pig 0.17.0已经停止维护多年、完全过时，推荐以下几种当前Hadoop生态的主流方案来实现BigDecimal类型相加：

方案1：自定义MapReduce Writable处理BigDecimal #

直接基于Hadoop原生API实现，不依赖任何过时组件：

• 自定义BigDecimalWritable类实现Writable接口，完成BigDecimal的序列化/反序列化：

import java.io.DataInput;
import java.io.DataOutput;
import java.io.IOException;
import java.math.BigDecimal;
import org.apache.hadoop.io.Writable;

public class BigDecimalWritable implements Writable {
    private BigDecimal value;

    public BigDecimalWritable() {}

    public BigDecimalWritable(BigDecimal value) {
        this.value = value;
    }

    @Override
    public void write(DataOutput out) throws IOException {
        out.writeUTF(value.toString());
    }

    @Override
    public void readFields(DataInput in) throws IOException {
        this.value = new BigDecimal(in.readUTF());
    }

    public BigDecimal getValue() {
        return value;
    }

    public void setValue(BigDecimal value) {
        this.value = value;
    }
}

• 在Reduce阶段对同Key下的所有BigDecimal值执行相加：

// Reduce示例逻辑
@Override
protected void reduce(Text key, Iterable<BigDecimalWritable> values, Context context) throws IOException, InterruptedException {
    BigDecimal sum = BigDecimal.ZERO;
    for (BigDecimalWritable val : values) {
        sum = sum.add(val.getValue());
    }
    context.write(key, new BigDecimalWritable(sum));
}

方案2：使用Apache Hive简化处理 #

Hive原生支持DECIMAL类型，无需编写Java代码，直接通过HQL完成求和：

• 创建表时指定DECIMAL字段：

CREATE TABLE data_table (
    id STRING,
    amount DECIMAL(18,2)
)
ROW FORMAT DELIMITED FIELDS TERMINATED BY ','
STORED AS TEXTFILE;

• 执行分组求和：

SELECT id, SUM(amount) AS total_amount
FROM data_table
GROUP BY id;

方案3：使用Apache Spark高效处理 #

Spark是当前Hadoop生态的主流计算框架，对Decimal类型支持完善：

• 用Spark SQL实现求和逻辑：

import org.apache.spark.sql.SparkSession
import org.apache.spark.sql.types.{DecimalType, StringType, StructField, StructType}

object BigDecimalSum {
    def main(args: Array[String]): Unit = {
        val spark = SparkSession.builder()
            .appName("BigDecimalSum")
            .master("local[*]") // 生产环境移除该配置
            .getOrCreate()

        val schema = StructType(Seq(
            StructField("id", StringType),
            StructField("amount", DecimalType(18,2))
        ))

        val df = spark.read.schema(schema).csv("hdfs://path/to/your/data")
        df.groupBy("id").sum("amount").show()

        spark.stop()
    }
}

• 也可通过DataFrame API直接操作，灵活性更强。

以上三种方案均基于活跃维护的技术栈，完全可以替代过时的Apache Pig方案。

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