1. 为什么批量写入场景下 BulkLoad 比 Put 更值得选
做过 HBase 写入的同学大概都有过这样的经历：用Table.put()一批一批往表里灌数据，刚开始几十万条还行，数据量一上千万，RegionServer 的写缓存开始频繁 flush，compaction 队列越堆越长，写入速度断崖式下跌，严重的时候直接把 RegionServer 拖到响应超时。这不是代码写得不好，而是Put这条路径本身就要走 WAL、MemStore、flush、compaction 这一整套流程，数据量越大，这套流程的代价就越明显。
BulkLoad 走的是完全不同的路子。它的核心思路是绕过 HBase 的写路径，直接利用 HFile 的存储格式，在外部（通常是 Spark、MapReduce 任务里）生成好 HFile 文件，然后通过LoadIncrementalHFiles把这些文件"搬运"到对应的 Region 目录下。整个过程不写 WAL、不经过 MemStore，对 RegionServer 的冲击极小，写入吞吐能比普通 Put 高出几倍甚至一个数量级。对于历史数据迁移、离线大批量导入、T+1 数据同步这类场景，BulkLoad 几乎是标配方案。
但 BulkLoad 也不是没有门槛。它要求你生成的数据必须严格按照 HFile 的格式组织，rowkey 必须有序，列族必须和目标表一致，而且一旦涉及多列（尤其是多个列族、多个限定符），生成 HFile 的逻辑就会变得比较绕。Spark 2.4.6 搭配 HBase 1.3.1 这个组合在不少存量项目里还在用，网上关于"单列 BulkLoad"的示例很多，但"多列 BulkLoad"的完整可跑代码却相对零散。这篇内容就围绕这个具体场景，把从环境准备到多列 HFile 生成、再到加载验证的完整链路拆开讲清楚，顺带把我在实际项目里踩过的坑一并交代。
适合阅读的对象包括：正在做 HBase 离线批量导入的工程师、需要把 Spark 计算结果高效落库的数据开发、以及被 Put 写入性能折磨过想换方案的同学。即使你之前没接触过 BulkLoad，跟着思路走也能把多列场景跑通。
2. Spark 2.4.6 与 HBase 1.3.1 的依赖对齐与版本陷阱
2.1 版本组合的兼容性判断
Spark 2.4.6 是 Scala 2.11 和 2.12 双版本发布的，HBase 1.3.1 官方编译产物默认依赖的是 Scala 2.11（如果你用的是hbase-spark那套）。这里第一个坑就来了：如果你的 Spark 集群是 Scala 2.12 编译的，直接引 HBase 1.3.1 的hbase-spark会报NoSuchMethodError或者ClassNotFoundException，因为二进制不兼容。我的建议是，做 BulkLoad 时干脆不要依赖hbase-spark这个集成包，直接用 HBase 原生的hbase-client、hbase-server、hbase-common这几个 jar，自己写 HFile 生成逻辑。这样版本耦合度最低，Scala 2.11 还是 2.12 都不影响。
具体依赖坐标（Maven 为例）：
<dependency> <groupId>org.apache.hbase</groupId> <artifactId>hbase-client</artifactId> <version>1.3.1</version> </dependency> <dependency> <groupId>org.apache.hbase</groupId> <artifactId>hbase-server</artifactId> <version>1.3.1</version> </dependency> <dependency> <groupId>org.apache.hbase</groupId> <artifactId>hbase-common</artifactId> <version>1.3.1</version> </dependency>注意hbase-server这个包体积不小，而且它会传递依赖 Hadoop 的一堆东西。在 Spark 任务里，这些依赖最好用--jars或者spark-submit的--packages方式带进去，不要一股脑打进 uber jar，否则容易和 Spark 自带的 Hadoop 版本冲突。HBase 1.3.1 默认对应 Hadoop 2.5.1 编译，但实际生产集群往往是 Hadoop 2.7.x 甚至 2.8.x，运行时以集群的 Hadoop 为准，编译期用 provided 就行。
2.2 集群侧需要提前确认的三件事
第一，目标 HBase 表的列族必须提前建好。BulkLoad 不会帮你建列族，HFile 里的列族名如果和表定义对不上，加载阶段会直接失败。建表时列族名建议全小写，避免大小写带来的诡异问题。
第二，hbase-site.xml要能被 Spark 任务读到。BulkLoad 过程中需要访问 HDFS 上的 HBase 根目录，还要和 HMaster、RegionServer 通信拿 Region 分布信息。最稳妥的做法是把集群的hbase-site.xml和core-site.xml、hdfs-site.xml一起放到 Spark 的 classpath 里，或者用--files分发。我见过有人只放了hbase-site.xml，结果LoadIncrementalHFiles阶段找不到 HDFS 的 nameservice，报UnknownHostException，排查半天。
第三，确认 HDFS 上有一个可写的临时目录用来存放生成的 HFile。这个目录最好在目标表所在的 HDFS 集群上，避免跨集群搬运。目录权限要给到提交 Spark 任务的用户，否则LoadIncrementalHFiles读不到文件。
2.3 一个容易被忽略的序列化问题
Spark 里生成 HFile 时，HFileContext、Configuration这些对象如果被闭包捕获，可能触发序列化异常。我的做法是把所有 HBase 相关的配置在 Driver 端读好，通过广播变量分发，Executor 端只做纯粹的数据转换和 HFile 写入。另外，Configuration本身不是Serializable的，千万别直接塞进 RDD 的闭包里，要么用@transient，要么在 Executor 端重新HBaseConfiguration.create()。
3. 多列场景下 HFile 生成的核心逻辑拆解
3.1 单列和多列的本质区别在哪
单列 BulkLoad 的示例代码通常长这样：每个 rowkey 对应一个KeyValue，列族固定，限定符固定，直接往HFile.Writer里 append 就完事。但多列场景下，一个 rowkey 往往对应多个列（比如info:name、info:age、info:city），甚至跨多个列族（info:xxx和stat:yyy）。这时候你不能简单地循环 append，因为 HFile 要求同一个 rowkey 下的所有 KeyValue 必须连续且有序，否则加载后数据会错乱甚至丢失。
HFile 的排序规则是：先按 rowkey 字典序，rowkey 相同再按列族字典序，列族相同再按限定符字典序，最后按时间戳倒序。所以生成多列 HFile 时，你必须保证写入的 KeyValue 序列严格满足这个顺序。最省心的做法是：在 Spark 里先把数据按(rowkey, family, qualifier)排好序，再逐条构造 KeyValue 写入。如果数据量特别大，排序本身会带来 shuffle，这时候可以考虑用repartitionAndSortWithinPartitions做分区内排序，减少全局排序的开销。
3.2 KeyValue 构造的关键参数
构造一个KeyValue需要这些信息：rowkey 字节数组、列族字节数组、限定符字节数组、时间戳、value 字节数组。时间戳如果不指定，默认用HConstants.LATEST_TIMESTAMP，实际写入时会由 HBase 填充当前时间。但在 BulkLoad 场景下，我建议显式指定时间戳，尤其是做数据迁移时，保留原始业务时间戳能避免版本覆盖问题。
val kv = new KeyValue( Bytes.toBytes(rowkey), Bytes.toBytes(family), Bytes.toBytes(qualifier), timestamp, Bytes.toBytes(value) )这里有个细节：KeyValue的构造函数有好几个重载，别用错。带Type参数的那个是给 Delete 标记用的，普通写入用上面这个五参数版本即可。另外 value 如果是 null，要转成空字节数组Bytes.toBytes("")，直接传 null 会在写入阶段抛 NPE。
3.3 HFile.Writer 的配置与关闭
HFile.Writer的创建方式在 HBase 1.x 里推荐用HFile.getWriterFactory(conf, cacheConf)这套 API，而不是老的new HFile.Writer()。CacheConfig要显式传，否则可能因为 block cache 配置缺失导致写入异常。
val conf = HBaseConfiguration.create() val cacheConf = new CacheConfig(conf) val writer = HFile.getWriterFactory(conf, cacheConf) .withPath(new Path(hfilePath)) .withFileContext(new HFileContext()) .create()写完所有 KeyValue 后，务必调用writer.close()，否则 HFile 的尾部元数据不完整，加载时会报CorruptHFileException。我习惯用 try-finally 包起来，确保异常时也能关闭。还有一点，HFileContext里可以设置压缩算法（比如withCompression(Algorithm.SNAPPY)），如果目标表建表时指定了压缩，这里最好保持一致，不然加载后读取可能出问题。
3.4 多列写入的代码骨架
把上面的点串起来，一个 rowkey 多列的写入逻辑大概是这样：
def writeRow(writer: HFile.Writer, rowkey: String, columns: Map[String, String]): Unit = { val rk = Bytes.toBytes(rowkey) columns.toSeq .sortBy { case (qualifier, _) => qualifier } .foreach { case (qualifier, value) => val kv = new KeyValue( rk, Bytes.toBytes("info"), Bytes.toBytes(qualifier), System.currentTimeMillis(), Bytes.toBytes(value) ) writer.append(kv) } }注意sortBy那一步，它保证了同一个 rowkey 下限定符有序。如果跨列族，排序键要改成(family, qualifier)的元组。这个排序看起来简单，但漏掉它是我见过最多的错误来源，加载后查询发现某些列"丢了"，其实就是顺序不对导致 HFile 索引错乱。
4. 从 Spark RDD 到 HFile 的完整落地流程
4.1 数据准备与分区策略
假设我们有一份用户画像数据，rowkey 是用户 ID，列包括姓名、年龄、城市、注册时间，都放在info列族下。数据源可能是 Hive 表、Parquet 文件或者上游 Kafka 落盘的文件。用 Spark 读进来后，第一步是确定分区数。分区数直接决定了最终生成多少个 HFile 文件，也影响加载时的并行度。
分区数太少，单个 HFile 过大，加载时一个 Region 要吞一个大文件，可能触发 split；分区数太多，小文件一堆，HFile 的索引开销和加载时的 RPC 次数都会上升。我的经验值是：目标表有多少个 Region，就设多少个分区，或者略多一点。比如表预分区了 32 个 Region，那 Spark 分区设 32 到 64 之间比较合适。可以用df.rdd.getNumPartitions看当前分区，再用repartition调整。
4.2 每个分区生成独立 HFile
BulkLoad 要求每个分区输出到一个独立的目录，目录下是该分区对应的 HFile。用mapPartitionsWithIndex可以拿到分区编号，据此拼出唯一的输出路径：
val hfileBasePath = "/tmp/bulkload/user_profile" rdd.mapPartitionsWithIndex { (index, iter) => val conf = HBaseConfiguration.create() val cacheConf = new CacheConfig(conf) val partitionPath = new Path(hfileBasePath + "/" + index) val writer = HFile.getWriterFactory(conf, cacheConf) .withPath(partitionPath) .withFileContext(new HFileContext()) .create() try { iter.foreach { record => // 构造 KeyValue 并 append } } finally { writer.close() } Iterator.empty }.count()这里用count()触发实际执行，因为mapPartitionsWithIndex是惰性的，不触发 action 的话 HFile 根本不会生成。用count()而不是collect()是为了避免把数据拉回 Driver。
4.3 分区内排序的必要性
上面这段代码有个隐患：iter里的数据顺序是不确定的，如果同一个 rowkey 的多条记录分散在不同位置，写入的 KeyValue 就不满足 HFile 的排序要求。解决办法是在mapPartitionsWithIndex内部先把迭代器转成 List，按 rowkey 排序后再写：
val sorted = iter.toList.sortBy(_._1) sorted.foreach { case (rowkey, columns) => writeRow(writer, rowkey, columns) }这样做的前提是单个分区的数据量不能太大，否则toList会把分区数据全加载到内存，有 OOM 风险。如果单分区数据量确实很大，就得用外部排序或者换用repartitionAndSortWithinPartitions在 shuffle 阶段完成排序。判断标准很简单：单分区数据量乘以单条记录大小，如果超过 Executor 内存的 1/3，就别用toList。
4.4 加载阶段的 LoadIncrementalHFiles
HFile 生成完毕后，用LoadIncrementalHFiles把它们搬到 HBase 表里：
val loader = new LoadIncrementalHFiles(conf) loader.doBulkLoad(new Path(hfileBasePath), admin, table, regionLocator)admin是HBaseAdmin实例，table是Table实例，regionLocator是RegionLocator。这三个对象都要在 Driver 端创建，用完记得关闭。doBulkLoad内部会遍历目录下的所有 HFile，根据 rowkey 范围匹配到对应的 Region，然后调用 HDFS 的 rename 把文件移过去。如果某个 HFile 的 rowkey 范围跨越了多个 Region，加载会失败，报IOException: Trying to load hfile that does not fit in region。这也是为什么前面强调分区策略要和 Region 分布对齐。
5. 实测中那些让人抓头的报错与排查路径
5.1 报错一：列族不匹配导致的加载失败
现象是doBulkLoad阶段抛IOException: HFile has a different column family。原因通常是 HFile 里写的列族名和表定义的列族名不一致，比如 HFile 里写的是INFO，表里建的是info。HBase 的列族名是大小写敏感的，别指望它帮你自动转换。排查方法很简单，用hbase org.apache.hadoop.hbase.io.hfile.HFile -m -f <hfile路径>看 HFile 的元数据，里面会列出列族名，和describe '表名'的输出对比一下就知道。
5.2 报错二：rowkey 越界
报错信息类似Trying to load hfile that does not fit in region。这个问题的根源是 HFile 的 rowkey 范围超出了目标 Region 的范围。常见于表做了预分区，但 Spark 生成 HFile 时没有按预分区的边界来切分数据。解决办法有两个：一是让 Spark 的分区逻辑和预分区边界对齐，用HFileOutputFormat2.configureIncrementalLoad那套（不过那是 MapReduce 的 API）；二是干脆不预分区，让 HBase 自己 split，但这样加载时可能触发大量 split，影响在线业务。我一般推荐第一种，虽然麻烦点，但可控。
5.3 报错三：时间戳导致的版本覆盖
这个坑比较隐蔽。如果 HFile 里所有 KeyValue 的时间戳都是同一个值（比如都用System.currentTimeMillis()在 Driver 端取了一次），那么同一个 rowkey 同一列的多条记录会互相覆盖，最后只剩一条。正确做法是每条记录用各自的时间戳，或者至少保证同一 rowkey 同一列只写一条。做数据迁移时，时间戳一定要从原始数据里带过来，别自己生成。
5.4 报错四：Executor 端找不到 HBase 配置
现象是 Executor 日志里报java.io.FileNotFoundException: hbase-site.xml。这是因为hbase-site.xml只在 Driver 端存在，没有分发到 Executor。解决办法是在spark-submit时加--files /path/to/hbase-site.xml，或者在代码里用sparkContext.addFile分发。更彻底的做法是把 HBase 配置打进 jar 包，但这样不利于配置变更，不推荐。
5.5 一个排查思路的总结
遇到 BulkLoad 失败，我的排查顺序是：先看 HFile 本身有没有问题（用 HFile 工具看元数据），再看 HFile 的 rowkey 范围是否匹配 Region，然后看列族名是否一致，最后看时间戳和版本。这四步能覆盖 90% 以上的问题。剩下的 10% 往往是环境问题，比如 HDFS 权限、网络连通性、jar 包冲突，这些就得具体问题具体分析了。
6. 性能调优与生产环境的几点经验
6.1 压缩算法的选择
HFile 支持多种压缩算法，Snappy 在压缩比和 CPU 开销之间比较平衡，是大多数场景的首选。如果对存储空间特别敏感，可以用 GZ，但写入和读取的 CPU 开销会明显上升。LZO 和 LZ4 也可以，但需要额外安装 native 库，部署麻烦。我的建议是：在线查询为主的表用 Snappy，冷数据归档用 GZ。注意 HFile 的压缩算法要和表定义的压缩算法一致，否则加载后读取会报错。
6.2 单文件大小的控制
HFile 文件不是越大越好。单个 HFile 过大，加载时一个 Region 要处理一个大文件，可能触发 Region split，split 过程中 Region 不可用，影响在线业务。一般来说，单个 HFile 控制在 1GB 到 2GB 之间比较合适。如果发现某个分区的 HFile 特别大，说明数据分布不均，需要调整分区策略或者加盐。
6.3 加载时机的选择
BulkLoad 虽然对 RegionServer 冲击小，但加载过程中还是会占用 HDFS 的带宽和 NameNode 的元数据操作。生产环境建议在业务低峰期执行，比如凌晨。另外，加载前最好确认目标表没有正在进行的 major compaction，否则可能因为文件移动导致 compaction 失败。
6.4 监控与验证
加载完成后，别急着收工。用count或者scan抽样验证数据是否完整，重点检查多列场景下每一列是否都写进去了。我习惯写一个校验脚本，对比源数据的行数和目标表的行数，再随机抽几条记录对比每一列的值。这一步能发现很多"看起来成功实际丢数据"的问题。
6.5 一个关于内存的提醒
生成 HFile 的 Executor 内存要留足。HFile.Writer内部有 block 缓存，默认 64KB 一个 block，如果单分区数据量大，block 数量多，内存占用会上升。可以通过HFileContext的withBlocksize调整 block 大小，但别调太小，否则索引膨胀。另外，CacheConfig里的blockCache在写入场景下其实用不到，可以设小一点，把内存留给数据本身。
7. 多列 BulkLoad 的边界与替代方案思考
多列 BulkLoad 能解决大部分离线批量导入的问题，但它不是万能的。如果你的数据需要频繁更新（同一 rowkey 同一列反复写），BulkLoad 就不太合适，因为它本质上是追加文件，频繁加载会产生大量小文件，反而拖慢读取。这种场景还是得用 Put，或者考虑 HBase 的Increment、Append接口。
另外，如果列的数量是动态的（比如用户标签，每个用户的标签数量不固定），生成 HFile 时排序和内存管理会更复杂。这时候可以考虑把动态列序列化成 JSON 存到一个列里，牺牲一点查询灵活性换取写入的简单性。具体怎么选，取决于你的查询模式：如果经常按单个标签过滤，那还是得展开成多列；如果只是整体读取，序列化存储更省事。
还有一个替代思路是用 HBase 的Import工具配合Export快照，但那是整表级别的操作，不适合增量导入。BulkLoad 在增量场景下的灵活性是它最大的优势。
我在实际项目里用这套方案处理过单表几十亿行的导入，Spark 任务跑了大概四十分钟生成 HFile，加载用了不到十分钟，全程对在线业务几乎没有影响。对比之前用 Put 写入动辄几个小时的耗时，提升非常明显。当然，前期在版本对齐和排序逻辑上花的时间也不少，但一次调通之后，后续的导入任务就是改改路径和字段映射的事，边际成本很低。如果你也在被 HBase 批量写入的性能问题困扰，不妨把 BulkLoad 这条路走一遍，踩过前面说的那几个坑之后，后面就是一马平川。
