SparkOnHBase报序列化错误：需用JavaSerializer但Result不可序列化

我之前处理过类似的场景，这个问题的核心原因很明确：org.apache.hadoop.hbase.client.Result类并没有实现Java标准的Serializable接口，而你配置的JavaSerializer只能序列化实现了该接口的对象——这也是切换到KryoSerializer就能正常运行的原因，因为Kryo支持序列化未实现Serializable的类（只要正确注册对应的类）。

既然业务要求必须使用JavaSerializer，咱们可以从以下几个可行的方向解决：

方案1：将Result转换为自定义可序列化POJO/集合 #

这是最直观且易维护的方案：直接从Result中提取你需要的业务数据，封装到自己定义的实现了Serializable接口的POJO里，或者使用Spark自带的可序列化Tuple、HashMap等结构。

举个Java代码示例：

// 自定义可序列化的业务POJO
public class HBaseBusinessRecord implements Serializable {
    private static final long serialVersionUID = 1L; // 建议显式声明序列化ID
    private String rowKey;
    private String userNickname;
    private Long registerTime;

    // 构造器、getter、setter方法
    public HBaseBusinessRecord(String rowKey, String userNickname, Long registerTime) {
        this.rowKey = rowKey;
        this.userNickname = userNickname;
        this.registerTime = registerTime;
    }
}

// 在RDD转换过程中处理Result，生成可序列化对象
JavaRDD<HBaseBusinessRecord> businessRDD = hbaseRDD.map(result -> {
    String rowKey = Bytes.toString(result.getRow());
    // 从Result中提取指定列的数据
    byte[] nicknameBytes = result.getValue(Bytes.toBytes("info"), Bytes.toBytes("nickname"));
    String nickname = nicknameBytes != null ? Bytes.toString(nicknameBytes) : null;
    
    byte[] timeBytes = result.getValue(Bytes.toBytes("info"), Bytes.toBytes("register_time"));
    Long registerTime = timeBytes != null ? Bytes.toLong(timeBytes) : null;
    
    return new HBaseBusinessRecord(rowKey, nickname, registerTime);
});

这样处理后，RDD中传递的是可序列化的业务对象，JavaSerializer就能正常处理，不会再抛出序列化异常。

方案2：自定义Result的可序列化包装器 #

如果你的业务逻辑必须保留Result对象的结构，可以自己实现一个可序列化的包装类，通过手动序列化/反序列化Result的内部数据来绕过限制。

核心思路是利用Java序列化的writeObject和readObject方法，手动处理Result的内容（因为Result本身没实现Serializable，但它的内部数据可以通过API获取）：

public class SerializableResultWrapper implements Serializable {
    private static final long serialVersionUID = 1L;
    private transient Result result; // transient标记该字段不自动序列化

    public SerializableResultWrapper(Result result) {
        this.result = result;
    }

    public Result getResult() {
        return result;
    }

    // 手动实现序列化逻辑
    private void writeObject(ObjectOutputStream out) throws IOException {
        // 序列化rowkey
        out.writeObject(result.getRow());
        // 序列化Cell的数量
        List<Cell> cells = result.listCells();
        out.writeInt(cells.size());
        // 逐个序列化Cell的核心数据
        for (Cell cell : cells) {
            out.writeObject(CellUtil.cloneFamily(cell));
            out.writeObject(CellUtil.cloneQualifier(cell));
            out.writeObject(CellUtil.cloneValue(cell));
            out.writeLong(cell.getTimestamp());
        }
    }

    // 手动实现反序列化逻辑，重新构建Result
    private void readObject(ObjectInputStream in) throws IOException, ClassNotFoundException {
        byte[] rowKey = (byte[]) in.readObject();
        int cellCount = in.readInt();
        List<Cell> cells = new ArrayList<>(cellCount);
        
        for (int i = 0; i < cellCount; i++) {
            byte[] family = (byte[]) in.readObject();
            byte[] qualifier = (byte[]) in.readObject();
            byte[] value = (byte[]) in.readObject();
            long timestamp = in.readLong();
            
            // 根据提取的数据重新构建Cell（注意HBase版本差异，这里用KeyValue示例）
            Cell cell = new KeyValue(rowKey, family, qualifier, timestamp, value);
            cells.add(cell);
        }
        // 用Cell列表重新构建Result对象
        result = Result.create(cells);
    }
}

使用时只需要将Result包装起来：

JavaRDD<SerializableResultWrapper> wrappedRDD = hbaseRDD.map(result -> new SerializableResultWrapper(result));
// 后续需要使用Result时，调用getResult()方法即可
wrappedRDD.foreach(wrapper -> {
    Result result = wrapper.getResult();
    // 执行你的业务逻辑
});

方案3：避免在跨节点传递Result #

如果序列化异常是在Spark Action操作（比如collect()、saveAsTextFile()）中出现的，那可以考虑在Executor端就完成对Result的处理，不要把Result对象传递到Driver端或者其他节点。

比如，如果你只需要Result中的rowKey，就直接在map操作中提取rowKey，再进行后续操作：

// 错误示例：直接collect Result会把不可序列化对象传到Driver
List<Result> rawResults = hbaseRDD.collect();

// 正确示例：在Executor端提取数据，传递可序列化的字符串
List<String> rowKeys = hbaseRDD.map(result -> Bytes.toString(result.getRow())).collect();

这样既满足了业务需求，又避免了序列化问题。

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