1. 项目背景与核心需求拆解
1.1 为什么会有这个需求
手里有一批数据要进HBase，量还不小，每天增量在千万级别。最开始用的是HBase原生的Put API，单条或者批量提交，跑了一段时间发现两个问题：一是写入吞吐上不去，RegionServer的写压力很大，频繁触发flush和compaction；二是Spark任务本身跑得慢，大量时间花在RPC等待上。后来换成BulkLoad方式，直接把HFile生成好再加载进HBase，写入性能提升了一个数量级，而且对RegionServer的冲击小了很多。
但问题来了——业务方的数据不是单列族单列的简单结构，一张表有多个列族，每个列族下面还有多个列，而且不同行的列还不完全一样。Spark生成的RowKey和列值需要按照HBase的存储格式组织成HFile，这就涉及到HFileOutputFormat2的使用、KeyFamilyQualifier的构造、以及多列场景下的数据组装逻辑。Spark 2.4.6搭配HBase 1.3.1这个组合虽然不算新，但在不少存量项目中仍然在跑，网上能找到的完整示例要么是单列场景，要么版本对不上，踩坑的人不少。
这篇文章就把整个链路拆开，从数据准备、HFile生成、到最终BulkLoad加载，把多列场景下的关键细节讲清楚。适合正在用Spark做HBase批量写入的开发者，尤其是那些数据列不固定、需要动态组装多列的场景。
1.2 整体方案选型对比
在Spark里往HBase写数据，常见的有三种路子：
| 方案 | 写入方式 | 吞吐量 | 对集群影响 | 适用场景 |
|---|---|---|---|---|
| Put API | 逐条/批量RPC | 低 | 大 | 小批量、实时性要求高 |
| BulkLoad | 生成HFile后加载 | 高 | 小 | 大批量、离线场景 |
| Spark-HBase Connector | 封装Put | 中 | 中 | 快速开发、数据量中等 |
BulkLoad的核心优势在于绕过了RegionServer的写路径，直接生成HFile文件，然后通过LoadIncrementalHFiles工具将文件移动到HBase的存储目录下。这个过程不经过WAL，不触发flush，对集群的冲击极小。缺点是不够实时，适合T+1或者小时级别的离线批处理。
选Spark 2.4.6是因为项目环境已经固定，HBase 1.3.1也是存量集群的版本。这两个版本的API兼容性整体没问题，但有几个细节需要注意，后面会展开。
2. 核心细节解析与实操要点
2.1 HFileOutputFormat2的工作机制
HFileOutputFormat2是HBase提供的一个OutputFormat，专门用于在MapReduce或Spark任务中生成HFile文件。它的核心逻辑是：每个Reducer（在Spark中是每个Partition）输出一个HFile文件，文件内部按照RowKey、列族、列限定符、时间戳排序。
在Spark中使用时，需要将数据转换成(ImmutableBytesWritable, KeyValue)的键值对形式。其中ImmutableBytesWritable是RowKey，KeyValue包含了完整的单元格信息——行键、列族、列限定符、时间戳、值。
这里有个关键点：KeyValue的构造函数需要传入行键、列族、列限定符、时间戳和值。对于多列场景，每一列都需要构造一个独立的KeyValue对象。也就是说，如果一行有5列，就需要生成5个KeyValue，它们的RowKey相同，但列族和列限定符不同。
val keyValue = new KeyValue( rowKeyBytes, // 行键 familyBytes, // 列族 qualifierBytes, // 列限定符 timestamp, // 时间戳 valueBytes // 值 )2.2 多列场景下的数据组装
多列场景的难点在于：数据源的结构可能是嵌套的，比如一个JSON对象里面有多层字段，需要扁平化成HBase的列。另外，不同行的列可能不一样，有些行有某些列，有些行没有。HBase本身是稀疏存储的，不存在的列不会占用空间，所以不需要为缺失的列填充空值。
在Spark中，通常的做法是先把数据转换成RDD[(ImmutableBytesWritable, KeyValue)]或者RDD[(ImmutableBytesWritable, List[KeyValue])]。如果是后者，还需要用flatMap展开成前者。我一般倾向于直接生成扁平的KeyValue列表，因为HFileOutputFormat2要求的就是扁平的键值对。
val hfileRDD = sourceRDD.flatMap { row => val rowKey = buildRowKey(row) val rowKeyBytes = new ImmutableBytesWritable(Bytes.toBytes(rowKey)) val timestamp = System.currentTimeMillis() val keyValues = new ListBuffer[KeyValue]() // 列族1下的多列 keyValues += new KeyValue(rowKeyBytes.get(), Bytes.toBytes("cf1"), Bytes.toBytes("col1"), timestamp, Bytes.toBytes(row.col1)) keyValues += new KeyValue(rowKeyBytes.get(), Bytes.toBytes("cf1"), Bytes.toBytes("col2"), timestamp, Bytes.toBytes(row.col2)) // 列族2下的多列 keyValues += new KeyValue(rowKeyBytes.get(), Bytes.toBytes("cf2"), Bytes.toBytes("col3"), timestamp, Bytes.toBytes(row.col3)) keyValues.map(kv => (rowKeyBytes, kv)) }注意：
KeyValue的构造函数中，行键参数是byte[]，不是ImmutableBytesWritable。所以需要调用rowKeyBytes.get()获取底层的字节数组。这个细节很容易搞错，编译报错的时候可以检查一下。
2.3 列族与列限定符的映射策略
多列场景下，列族和列限定符的映射关系需要提前规划好。通常有两种方式：
第一种是硬编码映射，在代码里写死哪些字段对应哪个列族和列限定符。这种方式简单直接，但扩展性差，新增字段需要改代码重新发布。
第二种是配置化映射，把映射关系放在配置文件或者配置表中，代码读取配置动态生成KeyValue。这种方式灵活，适合字段经常变化的场景。
我一般推荐第二种，尤其是在字段不固定的业务场景下。可以用一个简单的Map结构来维护映射关系：
val columnMapping = Map( "col1" -> ("cf1", "qualifier1"), "col2" -> ("cf1", "qualifier2"), "col3" -> ("cf2", "qualifier3") )然后在遍历数据字段时，根据字段名查找对应的列族和列限定符。如果某个字段在映射表中不存在，可以选择跳过或者记录日志。
2.4 时间戳的处理
HBase的每个单元格都有一个时间戳，默认是写入时的系统时间。在BulkLoad场景下，时间戳是在生成KeyValue时指定的。如果业务上需要保留数据本身的时间属性，比如数据产生的时间，那么应该用数据中的时间字段作为时间戳，而不是用当前系统时间。
val timestamp = row.eventTime // 使用数据中的时间字段如果不需要保留数据时间，直接用System.currentTimeMillis()也可以。但要注意，同一行的多个列如果时间戳不一致，在HBase中会表现为不同版本的数据。所以同一行的所有列最好使用同一个时间戳，避免版本混乱。
3. 实操过程与核心环节实现
3.1 环境准备与依赖配置
先确认Spark和HBase的版本，以及它们之间的依赖关系。Spark 2.4.6默认依赖的HBase版本是1.2.6左右，但我们需要用1.3.1，所以需要在项目中显式引入HBase 1.3.1的依赖，并排除Spark自带的HBase依赖。
Maven的pom.xml配置如下：
<properties> <spark.version>2.4.6</spark.version> <hbase.version>1.3.1</hbase.version> </properties> <dependencies> <dependency> <groupId>org.apache.spark</groupId> <artifactId>spark-core_2.11</artifactId> <version>${spark.version}</version> <exclusions> <exclusion> <groupId>org.apache.hbase</groupId> <artifactId>hbase-client</artifactId> </exclusion> <exclusion> <groupId>org.apache.hbase</groupId> <artifactId>hbase-server</artifactId> </exclusion> </exclusions> </dependency> <dependency> <groupId>org.apache.hbase</groupId> <artifactId>hbase-client</artifactId> <version>${hbase.version}</version> </dependency> <dependency> <groupId>org.apache.hbase</groupId> <artifactId>hbase-server</artifactId> <version>${hbase.version}</version> </dependency> <dependency> <groupId>org.apache.hbase</groupId> <artifactId>hbase-mapreduce</artifactId> <version>${hbase.version}</version> </dependency> </dependencies>提示：
hbase-mapreduce这个包在HBase 1.3.1中是必须的，因为HFileOutputFormat2和LoadIncrementalHFiles都在这个包里。有些教程只引了hbase-client和hbase-server，结果运行时报ClassNotFoundException，就是漏了这个包。
3.2 HBase表结构设计
在生成HFile之前，需要先确认HBase表的结构。假设我们要写入的表叫test_table，有两个列族cf1和cf2，每个列族下有多个列限定符。
建表语句如下：
create 'test_table', {NAME => 'cf1', VERSIONS => 1, BLOOMFILTER => 'ROW'}, {NAME => 'cf2', VERSIONS => 1, BLOOMFILTER => 'ROW'}注意：
VERSIONS => 1表示只保留一个版本的数据。如果业务上需要保留多个版本，可以调大这个值。但在BulkLoad场景下，如果时间戳相同，多个版本会互相覆盖，需要特别注意。
3.3 Spark任务的核心代码
完整的Spark任务代码分为几个部分：读取数据源、构建RowKey、组装KeyValue、生成HFile、加载到HBase。
import org.apache.hadoop.fs.Path import org.apache.hadoop.hbase.{HBaseConfiguration, KeyValue, TableName} import org.apache.hadoop.hbase.client.{ConnectionFactory, HTable} import org.apache.hadoop.hbase.io.ImmutableBytesWritable import org.apache.hadoop.hbase.mapreduce.{HFileOutputFormat2, LoadIncrementalHFiles} import org.apache.hadoop.hbase.util.Bytes import org.apache.spark.sql.SparkSession import scala.collection.mutable.ListBuffer object SparkBulkLoadMultiColumn { def main(args: Array[String]): Unit = { val spark = SparkSession.builder() .appName("SparkBulkLoadMultiColumn") .enableHiveSupport() .getOrCreate() val hbaseConf = HBaseConfiguration.create() hbaseConf.set("hbase.zookeeper.quorum", "zk1,zk2,zk3") hbaseConf.set("hbase.zookeeper.property.clientPort", "2181") val tableName = "test_table" val hfilePath = "/tmp/hfile_output" // 读取数据源 val sourceDF = spark.sql("SELECT * FROM source_table") // 构建RDD[(ImmutableBytesWritable, KeyValue)] val hfileRDD = sourceDF.rdd.flatMap { row => val rowKey = row.getAs[String]("id") val rowKeyBytes = Bytes.toBytes(rowKey) val immutableRowKey = new ImmutableBytesWritable(rowKeyBytes) val timestamp = System.currentTimeMillis() val keyValues = new ListBuffer[(ImmutableBytesWritable, KeyValue)]() // 列族cf1下的列 val cf1 = Bytes.toBytes("cf1") keyValues += ((immutableRowKey, new KeyValue(rowKeyBytes, cf1, Bytes.toBytes("name"), timestamp, Bytes.toBytes(row.getAs[String]("name"))))) keyValues += ((immutableRowKey, new KeyValue(rowKeyBytes, cf1, Bytes.toBytes("age"), timestamp, Bytes.toBytes(row.getAs[Int]("age").toString)))) // 列族cf2下的列 val cf2 = Bytes.toBytes("cf2") keyValues += ((immutableRowKey, new KeyValue(rowKeyBytes, cf2, Bytes.toBytes("address"), timestamp, Bytes.toBytes(row.getAs[String]("address"))))) keyValues += ((immutableRowKey, new KeyValue(rowKeyBytes, cf2, Bytes.toBytes("phone"), timestamp, Bytes.toBytes(row.getAs[String]("phone"))))) keyValues.toList } // 配置HFileOutputFormat2 val table = ConnectionFactory.createConnection(hbaseConf) .getTable(TableName.valueOf(tableName)) val regionLocator = ConnectionFactory.createConnection(hbaseConf) .getRegionLocator(TableName.valueOf(tableName)) HFileOutputFormat2.configureIncrementalLoad( spark.sparkContext.hadoopConfiguration, table, regionLocator ) // 生成HFile hfileRDD.saveAsNewAPIHadoopFile( hfilePath, classOf[ImmutableBytesWritable], classOf[KeyValue], classOf[HFileOutputFormat2], hbaseConf ) // 加载HFile到HBase val loadIncrementalHFiles = new LoadIncrementalHFiles(hbaseConf) loadIncrementalHFiles.doBulkLoad( new Path(hfilePath), table.asInstanceOf[HTable] ) spark.stop() } }3.4 关键参数计算与选择
RowKey设计：RowKey的设计直接影响到HBase的读写性能。常见的策略有加盐、哈希、反转等。如果数据量不大，直接用业务主键也可以。但如果数据量大且写入集中，建议做散列处理，避免热点。
// 加盐示例：在RowKey前加两位随机数 val salt = (Math.abs(rowKey.hashCode) % 100).toString val saltedRowKey = salt + "_" + rowKeyHFile大小：HFile的大小由hbase.hregion.max.filesize参数控制，默认是10GB。在BulkLoad场景下，生成的HFile大小取决于每个Partition的数据量。如果单个HFile过大，加载时可能会触发Region分裂。可以通过调整Spark的Partition数量来控制每个HFile的大小。
// 调整Partition数量 val hfileRDD = sourceDF.rdd.repartition(100).flatMap { ... }时间戳精度：HBase的时间戳是毫秒级。如果同一行在同一毫秒内写入多次，后面的会覆盖前面的。在BulkLoad场景下，如果数据源中有重复的RowKey，需要提前去重或者使用不同的时间戳。
3.5 实操现场记录
在实际跑任务的时候，有几个环节需要特别关注：
第一步：确认HBase表存在且列族配置正确。如果表不存在，configureIncrementalLoad会报错。列族的VERSIONS配置也要注意，如果设置为1，那么相同RowKey相同列族相同列限定符的数据只会保留最新版本。
第二步：检查HFile输出目录。saveAsNewAPIHadoopFile会生成一个目录，里面包含_SUCCESS文件和多个HFile文件。如果目录已存在，任务会报错，需要提前删除。
hadoop fs -rm -r /tmp/hfile_output第三步：执行BulkLoad。doBulkLoad方法会将HFile文件移动到HBase的存储目录下。这个过程是原子的，要么全部成功，要么全部失败。如果失败，需要检查HBase的日志，常见原因是HFile的格式不匹配或者Region的边界不对。
第四步：验证数据。加载完成后，用HBase Shell扫描表，确认数据是否正确写入。
scan 'test_table', {LIMIT => 10}4. 常见问题与排查技巧实录
4.1 常见问题速查表
| 问题现象 | 可能原因 | 解决方法 |
|---|---|---|
| ClassNotFoundException: HFileOutputFormat2 | 缺少hbase-mapreduce依赖 | 添加hbase-mapreduce依赖 |
| IllegalArgumentException: Can't read hfile | HFile格式不匹配 | 确认HBase版本与生成HFile的版本一致 |
| RegionTooBusyException | Region写入压力大 | 调整RowKey散列，增加Region数量 |
| HFile加载后数据为空 | 时间戳或列族配置错误 | 检查KeyValue构造参数 |
| 任务OOM | Partition数据量过大 | 增加Partition数量，减少单Partition数据量 |
| BulkLoad失败：FileNotFoundException | HFile输出目录不存在 | 确认saveAsNewAPIHadoopFile执行成功 |
4.2 独家避坑技巧
坑一：HBase依赖冲突。Spark自带的HBase依赖版本和项目需要的版本不一致，会导致各种奇怪的错误。最稳妥的做法是在pom.xml中排除Spark自带的HBase依赖，显式引入项目需要的版本。如果用的是SBT，也需要做类似的排除。
坑二：KeyValue的RowKey参数。KeyValue的构造函数第一个参数是byte[]，不是ImmutableBytesWritable。很多示例代码里写的是rowKeyBytes.get()，但如果rowKeyBytes是ImmutableBytesWritable类型，需要先调用get()方法获取字节数组。这个细节在编译时可能不会报错，但运行时会出问题。
坑三：时间戳不一致导致数据覆盖。如果同一行的不同列使用了不同的时间戳，在HBase中会表现为不同版本。如果表的VERSIONS设置为1，那么只有最新时间戳的列会被保留，其他列会丢失。所以同一行的所有列必须使用同一个时间戳。
坑四：HFile输出目录权限问题。BulkLoad需要将HFile文件从输出目录移动到HBase的存储目录，这涉及到HDFS的权限。如果Spark任务使用的用户没有权限访问HBase的存储目录，BulkLoad会失败。解决方法是确保Spark任务使用的用户有足够的权限，或者使用HBase的管理员用户执行BulkLoad。
坑五：Region边界不对。如果HBase表的预分区和HFile中的RowKey范围不匹配，BulkLoad时会出现RegionSplit异常。解决方法是提前做好预分区，确保HFile中的RowKey落在正确的Region范围内。
4.3 性能调优建议
调整Partition数量：Partition数量决定了生成的HFile文件数量。Partition太少，单个HFile过大，加载时可能触发Region分裂；Partition太多，生成大量小HFile，加载效率低。一般建议每个Partition处理100MB到500MB的数据。
使用Snappy压缩：HFile支持多种压缩算法，Snappy在压缩比和速度之间取得了较好的平衡。可以在建表时指定压缩算法：
create 'test_table', {NAME => 'cf1', COMPRESSION => 'SNAPPY'}, {NAME => 'cf2', COMPRESSION => 'SNAPPY'}预分区：在创建表时预先划分好Region，避免BulkLoad时自动分裂。预分区的数量根据数据量和RowKey分布来定，一般建议每个Region的大小在1GB到5GB之间。
create 'test_table', {NAME => 'cf1', COMPRESSION => 'SNAPPY'}, {NAME => 'cf2', COMPRESSION => 'SNAPPY'}, SPLITS => ['1', '2', '3', '4', '5', '6', '7', '8', '9']批量提交：如果数据量特别大，可以分批次执行BulkLoad，避免单次加载过多HFile导致RegionServer压力过大。
5. 多列场景下的扩展思考
5.1 动态列的處理
实际业务中，列的数量和名称可能是不固定的。比如用户画像数据，每个用户的标签数量不一样，标签名称也不一样。这种情况下，需要在代码中动态遍历数据字段，根据字段名生成对应的列限定符。
val dynamicColumns = row.getValuesMap(row.schema.fieldNames) dynamicColumns.foreach { case (fieldName, fieldValue) => val (family, qualifier) = columnMapping.getOrElse(fieldName, ("cf_default", fieldName)) keyValues += ((immutableRowKey, new KeyValue( rowKeyBytes, Bytes.toBytes(family), Bytes.toBytes(qualifier), timestamp, Bytes.toBytes(fieldValue.toString) ))) }这种方式的好处是扩展性强，新增字段不需要改代码。但需要注意字段名到列族和列限定符的映射关系要维护好，避免出现列族或列限定符不存在的情况。
5.2 多列族写入的注意事项
HBase的列族数量不宜过多，一般建议不超过3个。列族过多会导致MemStore和StoreFile数量增加，影响性能。如果业务上确实需要多个列族，建议把访问模式相似的列放在同一个列族下。
在BulkLoad场景下，多个列族的HFile会分别生成，加载时也会分别处理。如果某个列族的数据量特别大，可以考虑单独做BulkLoad，避免单个任务处理过多数据。
5.3 数据一致性保障
BulkLoad的一个潜在问题是数据一致性。因为BulkLoad绕过了WAL，如果加载过程中出现故障，可能会导致部分数据丢失。为了保证数据一致性，可以采取以下措施：
第一，在BulkLoad之前，先确认HFile文件已经完整生成，并且_SUCCESS文件存在。
第二，BulkLoad完成后，用HBase的hbck工具检查表的完整性。
第三，如果业务上对数据一致性要求极高，可以在BulkLoad之后做一次数据校验，对比源数据和HBase中的数据。
hbase hbck -details test_table5.4 与其他写入方式的对比
BulkLoad虽然性能好，但并不是所有场景都适用。如果数据量不大，或者对实时性要求高，用Put API或者Spark-HBase Connector可能更合适。BulkLoad更适合T+1的离线批处理场景，数据量大、对实时性要求不高。
另外，BulkLoad生成的HFile文件需要占用HDFS空间，如果数据量特别大，需要考虑HDFS的存储容量。而Put API是直接写入HBase的，不需要额外的HDFS空间。
我在实际项目中的体会是，BulkLoad和Put API可以结合使用。历史数据用BulkLoad批量导入，增量数据用Put API实时写入。这样既能保证历史数据的导入效率，又能满足增量数据的实时性要求。
最后再分享一个小技巧：如果BulkLoad过程中出现RegionTooBusyException，可以适当降低BulkLoad的并发度，或者调整HBase的hbase.regionserver.handler.count参数，增加RegionServer的处理线程数。这个参数默认是30，可以根据实际情况调大到50或者100。
