如何使用Spark多线程写入HBase及单Executor写入过慢问题优化

• 首先确认上游DataFrame的分区数：你通过SQL查询得到的df如果本身只有1个分区，后续转RDD做mapPartitions也只会有1个任务，对应只用1个核心执行。你可以在代码中添加println(df.rdd.getNumPartitions)打印分区数验证，如果分区数确实为1，需要先通过repartition(并行度)调整分区，并行度建议设置为可用核心数的1~2倍。
• 确认Spark任务的资源申请配置：检查spark-submit提交时的--num-executors、--executor-cores参数是否设置过小，如果仅申请了1个Executor、1个核心，自然只会有单核心执行任务。
• 确认HBase表的Region数量：如果HBase表未做预分区，所有数据都写入同一个Region，会导致单Region热点，Spark侧即使有多个分区也会因为Region锁的限制无法并行写入，表现为只有少量甚至1个任务在执行。

代码层面优化 #

• 减少不必要的对象创建：你当前在每条数据处理时都新建ImmutableBytesWritable对象，可以改成在mapPartitions中每个分区共用一个对象，降低GC开销：

val result = df.rdd.mapPartitions(par=>{
  val immutableBytesWritable = new ImmutableBytesWritable()
  par.map(row=>{
    // 原有行处理逻辑不变
    (immutableBytesWritable, put)
  })
})

• 开启HBase客户端批量写缓存：在hbaseConf中添加以下配置，开启批量提交减少RPC请求次数，大幅提升写入效率：

// 写缓冲大小设为2M，可根据实际硬件配置调整
hbaseConf.set("hbase.client.write.buffer", "2097152") 
hbaseConf.set(TableOutputFormat.WRITE_BUFFER_SIZE, "2097152")

• 优化字段序列化逻辑：当前代码将所有字段值转成String后再序列化存入HBase，会增大数据体积，建议根据字段实际类型选择对应的序列化方式，比如数值型字段直接调用Bytes.toBytes(数值)，不要转String。

集群层面优化 #

• HBase表提前预分区：建表时根据rowkey的分布规则提前划分多个Region，避免数据倾斜和单Region热点，保证Spark多个分区可以并行写入不同的Region。
• 调整写入并行度：在调用saveAsNewAPIHadoopDataset之前，可以根据HBase表的Region数量，对RDD调用repartition(Region数 * 2)调整分区数，充分利用集群资源。
• 调整HBase客户端参数：适当调小hbase.client.retries.number（默认31次，可调整为3~5次），调整hbase.rpc.timeout到合适值，避免异常重试占用过多时间。

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