Spark写入HBase数据不完整问题求助

看起来你遇到了Spark写入HBase时大量数据丢失+数据乱序的问题，而写入MySQL正常，说明数据从Kafka到Spark的消费环节是没问题的，核心问题出在Spark到HBase的写入逻辑上。下面分点拆解原因和解决办法：

1. 最关键原因：RowKey重复导致数据覆盖 #

你当前用System.currentTimeMillis()作为HBase的RowKey，这个值的精度是毫秒级。Spark是并行处理数据的，同一毫秒内会有大量记录生成完全相同的RowKey——HBase的Put操作是幂等的，后写入的同RowKey数据会直接覆盖之前的，这就是为什么100条只剩36条、3万条只剩4000条的核心原因（剩下的都是不同毫秒生成的唯一RowKey）。

解决办法：生成全局唯一的RowKey #

推荐两种方案：

• 方案1：用UUID生成唯一RowKey（最简单，无需依赖Kafka元数据）

  // 替换原来的RowKey生成逻辑
  Put put = new Put(Bytes.toBytes(UUID.randomUUID().toString()));
  

• 方案2：结合Kafka元数据生成有序RowKey（如果需要数据按消费顺序存储）
  如果你是从Kafka消费ConsumerRecord，可以用时间戳+分区ID+偏移量组合，既保证唯一，又能保证分区内的顺序：

  // 假设你的rdd是JavaRDD<ConsumerRecord<String, String>>
  JavaPairRDD<ImmutableBytesWritable, Put> hbasePuts = rowRDD.mapToPair(row -> {
      long timestamp = System.currentTimeMillis();
      int partition = row.partition();
      long offset = row.offset();
      String rowKey = timestamp + "_" + partition + "_" + offset;
      Put put = new Put(Bytes.toBytes(rowKey));
      // 后续的addImmutable逻辑不变
      put.addImmutable(Bytes.toBytes("time"), Bytes.toBytes("col1"), Bytes.toBytes(row.getTime()));
      put.addImmutable(Bytes.toBytes("time_taken"), Bytes.toBytes("col2"), Bytes.toBytes(row.getTime_taken()));
      put.addImmutable(Bytes.toBytes("ip"), Bytes.toBytes("col3"), Bytes.toBytes(row.getIp()));
      return new Tuple2<>(new ImmutableBytesWritable(), put);
  });
  

2. 次要问题：foreachRDD中重复创建SparkSession #

你在foreachRDD的闭包里每次都创建SparkSession，这是完全不必要的——SparkSession是Driver级别的单例对象，重复创建会浪费资源，甚至可能引发线程安全问题。

修改方式：把SparkSession移到Driver端初始化 #

// 在StreamingContext初始化之后，foreachRDD之前创建SparkSession
SparkSession spark = SparkSession.builder().config(ssc.sparkContext().getConf()).getOrCreate();

lines.foreachRDD((rdd, time)-> {
    // 直接复用外面的spark实例，不需要重新创建
    JavaRDD<Log> rowRDD = rdd.map(line -> {
        String[] logLine = line.split(" +");
        Log record = new Log();
        record.setTime(logLine[0]);
        record.setTime_taken(logLine[1]);
        record.setIp(logLine[2]);
        return record;
    });
    saveToHBase(rowRDD, newAPIJobConfiguration1.getConfiguration());
});

3. 额外优化：HBase写入的稳定性配置 #

可以调整HBase客户端的写入配置，避免因缓冲或超时导致的数据丢失：

// 在初始化Configuration时添加以下配置
conf.set("hbase.client.write.buffer", "2097152"); // 设置2MB的写入缓冲，根据数据量调整
conf.set("hbase.client.retries.number", "3"); // 设置重试次数
conf.set("hbase.client.operation.timeout", "60000"); // 设置操作超时时间

4. 数据乱序的解决 #

你提到HBase中的记录无连续顺序，这是因为Spark并行处理不同分区的RDD，写入HBase的顺序是不确定的，再加上用毫秒级时间戳当RowKey，同一毫秒的RowKey会被覆盖，自然没有顺序。用上面的**方案2（时间戳+分区+偏移量）**就能保证分区内的数据是按消费顺序存储的，全局也能保持大致的时间顺序。

修改后的saveToHBase方法示例 #

public static void saveToHBase(JavaRDD<Log> rowRDD, Configuration conf) throws IOException {
    JavaPairRDD<ImmutableBytesWritable, Put> hbasePuts = rowRDD.mapToPair(row -> {
        // 使用UUID生成唯一RowKey
        String rowKey = UUID.randomUUID().toString();
        Put put = new Put(Bytes.toBytes(rowKey));
        put.addImmutable(Bytes.toBytes("time"), Bytes.toBytes("col1"), Bytes.toBytes(row.getTime()));
        put.addImmutable(Bytes.toBytes("time_taken"), Bytes.toBytes("col2"), Bytes.toBytes(row.getTime_taken()));
        put.addImmutable(Bytes.toBytes("ip"), Bytes.toBytes("col3"), Bytes.toBytes(row.getIp()));
        return new Tuple2<>(new ImmutableBytesWritable(), put);
    });
    hbasePuts.saveAsNewAPIHadoopDataset(conf);
}

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