企业官网建设流程全解析

1. 为什么要在 Spark 里走 BulkLoad 这条路

1.1 从一次写入瓶颈说起

早些年做数据平台的时候，遇到过一个很典型的需求：上游是每天几千万到上亿条的用户行为日志，经过 Spark 清洗、聚合之后，要落到 HBase 里供在线接口做点查。最开始用的是最朴素的方式——Table.put()一条条写，或者用saveAsNewAPIHadoopDataset走 TableOutputFormat。数据量小的时候还行，一旦单批次超过千万级，RegionServer 就开始扛不住，写请求排队、Region 频繁 split、compaction 压力陡增，整个集群的读写延迟都被拖垮。

后来换成了 BulkLoad 方案，同样的数据量，写入时间从原来的几小时压缩到十几分钟，而且对 RegionServer 的在线读写几乎零影响。这个差距不是一点半点，是数量级的。所以只要你的场景是"大批量、离线、一次性灌入 HBase"，BulkLoad 基本就是首选方案。

这篇内容就围绕一个具体的技术组合展开：Spark 2.4.6 + HBase 1.3.1，用 BulkLoad 方式把数据写入多列族/多列限定符的表。为什么强调"多列"？因为单列的场景网上例子一抓一大把，但真实业务里一张 HBase 表往往有多个列族、每个列族下十几个列，rowkey 设计、列族划分、HFile 生成时的列限定符处理，这些细节才是真正卡人的地方。

1.2 BulkLoad 到底在干什么

先把这个机制讲透，不然后面踩坑都不知道为什么。

HBase 的底层存储是 HFile，数据最终以 HFile 的形式落在 HDFS 上，由 RegionServer 加载并提供服务。正常写入路径是：客户端把数据发给 RegionServer，先写 WAL，再写 MemStore，MemStore 满了 flush 成 HFile。这条路径的问题在于，所有数据都要经过 RegionServer 这一层，网络、内存、磁盘 IO 全压在它身上。

BulkLoad 的思路是"绕过 RegionServer，直接生成 HFile"。具体分两步：

1. 生成 HFile：用 MapReduce（或者 Spark 借助 HBase 的HFileOutputFormat2）在 HDFS 上直接生成符合 HBase 格式的 HFile 文件，这一步完全不碰 RegionServer。
2. 加载 HFile：通过LoadIncrementalHFiles工具，把生成好的 HFile 移动到对应 Region 的目录下，然后通知 RegionServer 生效。这一步只是文件移动和元数据操作，非常轻量。

打个比方：正常写入像是你一件件把货搬进仓库，还要登记入库；BulkLoad 则是你在仓库外面把货按货架分好、打包成标准箱，然后直接整箱推进对应的货架位置。后者显然快得多，而且不占用仓库的作业通道。

1.3 多列场景下真正的难点在哪

单列族单列的场景，HFile 生成逻辑很简单，family和qualifier都是固定的。但多列场景下，问题就来了：

• 列族和列限定符是动态的：不同业务线的数据，列族可能不一样，列限定符更是千变万化，不能写死。
• rowkey 设计直接影响 HFile 的分布：如果 rowkey 设计得不好，生成的 HFile 会严重倾斜，加载时某个 Region 特别大，其他 Region 空着。
• 列族必须在建表时就存在：HFile 里如果出现了表里没有的列族，加载会直接失败。
• Spark 的 schema 和 HBase 的列结构要对齐：DataFrame 的列名怎么映射到 HBase 的family:qualifier，这是核心。

下面我会把整套流程拆开，从环境准备、rowkey 设计、HFile 生成、加载，到常见报错排查，一步步讲清楚。

2. 环境准备与依赖梳理

2.1 版本匹配这件事不能马虎

Spark 和 HBase 的版本兼容性是个老生常谈的坑。这里用的是 Spark 2.4.6 和 HBase 1.3.1，这个组合是经过验证能跑通的，但依赖包的引入方式有讲究。

核心依赖是hbase-server、hbase-client、hbase-common、hbase-mapreduce（或者老版本里的hbase-server里就包含了 HFileOutputFormat2）。在 Spark 2.4.6 里，通常用 Maven 管理依赖，pom 里大致是这样：

<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-mapreduce</artifactId> <version>1.3.1</version> </dependency>

注意：HBase 1.3.1 的hbase-mapreduce模块里才有HFileOutputFormat2和LoadIncrementalHFiles，别引错包。另外，HBase 依赖的 Hadoop 版本要和你集群的 Hadoop 版本对齐，否则会出现NoSuchMethodError这类运行时错误。

2.2 集群侧的前置条件

在动手写代码之前，集群侧有几件事必须先确认：

• HBase 表已经建好，并且列族已经创建。BulkLoad 不会自动建表，也不会自动建列族。如果 HFile 里出现了表里不存在的列族，加载阶段会报Do not have family xxx之类的错误。
• HDFS 上有可写的临时目录，用来存放生成的 HFile。这个目录最好和 HBase 的根目录在同一个 HDFS 集群，避免跨集群移动文件。
• Spark 任务有权限访问 HBase 的配置，通常是把hbase-site.xml和hdfs-site.xml、core-site.xml放到 Spark 的 classpath 里，或者通过--files分发。

我一般习惯把 HBase 的hbase-site.xml直接打进 Spark 应用的 jar 包里，或者放在$SPARK_HOME/conf下，这样最省事。如果集群是多租户环境，权限这块要提前和运维确认，否则加载阶段会因为 HDFS 目录权限被拒。

2.3 一个容易忽略的点：HFile 的临时目录清理

BulkLoad 生成的 HFile 是临时文件，加载完成后理论上可以删除，但实际生产中经常忘记清理，导致 HDFS 空间被慢慢吃掉。我的做法是在任务最后加一个清理步骤，或者用独立的定时任务清理超过一定时间的临时目录。这个细节后面在"避坑"部分还会展开。

3. rowkey 设计与列族规划

3.1 rowkey 决定了 HFile 的分布

BulkLoad 生成的 HFile 是按 rowkey 排序的，加载时 HBase 会根据 rowkey 的范围把 HFile 分配到对应的 Region。如果 rowkey 是单调递增的（比如时间戳前缀），那么所有新数据都会落到最后一个 Region，造成热点，加载时那个 Region 会特别慢，甚至超时。

常见的解法是加盐（salting）或者哈希前缀。比如原始 rowkey 是userId + timestamp，可以改成hash(userId) % N + userId + timestamp，N 是预分区数。这样数据会均匀分布到 N 个 Region。

但加盐有个副作用：范围查询会变复杂。如果你的业务既有批量写入又有范围扫描，rowkey 设计要权衡。我的经验是，如果查询主要是按 userId 点查，加盐完全没问题；如果还要按时间范围扫，那就要考虑用userId的哈希做前缀，时间放在后面，查询时按 userId 定位。

3.2 列族规划的几个原则

HBase 官方建议一张表的列族不要超过 2-3 个，因为列族是物理隔离的，每个列族有独立的 MemStore 和 HFile。列族太多会导致 flush 和 compaction 频繁，性能下降。

在多列场景下，我的建议是：

• 按访问模式划分列族：经常一起查询的列放同一个列族，比如"基本信息"一个列族，"扩展属性"一个列族。
• 列族名尽量短：列族名会存储在每个 KeyValue 里，太长会浪费空间。用f1、f2这种短名，或者用有意义的单字母。
• 列限定符可以动态：HBase 的列限定符是动态的，不需要预先定义，所以多列场景下 qualifier 可以灵活变化，但列族必须提前建好。

举个实际例子，一张用户画像表，可以这样设计：

这样划分之后，查询基础信息只扫b列族，查标签只扫t列族，IO 隔离得很清楚。

3.3 预分区不能省

建表时一定要预分区。如果不预分区，表初始只有一个 Region，BulkLoad 加载时所有 HFile 都往这一个 Region 塞，加载完再 split，这个过程非常慢，而且 split 期间读写都会受影响。

预分区的方式有两种：一是指定 split keys，二是用RegionSplitter工具。split keys 的选择要结合 rowkey 的分布，比如用哈希前缀的话，split keys 就是00、01、02... 这样均匀切分。

create 'user_profile', {NAME => 'b', VERSIONS => 1}, {NAME => 't', VERSIONS => 1}, {NAME => 's', VERSIONS => 1}, SPLITS => ['10', '20', '30', '40', '50', '60', '70', '80', '90']

这个建表语句里，SPLITS指定了 9 个 split key，加上默认的起止，一共 10 个 Region。具体分多少个，要看数据量和 Region 大小，一般单个 Region 控制在 10-20GB 比较合适。

4. Spark 侧生成 HFile 的核心实现

4.1 整体数据流

Spark 侧的任务可以概括为：读取源数据 -> 转换成(ImmutableBytesWritable, KeyValue)或(ImmutableBytesWritable, Put)的 RDD -> 用HFileOutputFormat2生成 HFile。

这里有个关键点：HFileOutputFormat2是 Hadoop 的 OutputFormat，Spark 通过saveAsNewAPIHadoopFile或者saveAsNewAPIHadoopDataset来调用它。在 Spark 2.4.6 里，推荐用saveAsNewAPIHadoopFile，因为它对 RDD 的类型要求更明确。

数据转换的核心是把业务字段映射成 HBase 的family:qualifier -> value。多列场景下，这个映射是动态的，通常用一个Map[String, String]来表示一行数据的所有列，key 是family:qualifier，value 是列值。

4.2 构造 Put 对象还是直接构造 KeyValue

两种方式都可以，但适用场景不同：

• 构造 Put 对象：更符合 HBase 的 API 习惯，Put.addColumn(family, qualifier, value)，然后HFileOutputFormat2内部会把 Put 转成 KeyValue。这种方式代码可读性好，适合列结构相对固定的场景。
• 直接构造 KeyValue：更底层，性能略好，但需要自己处理时间戳、KeyValue 的排序等细节。适合对性能极致追求、列结构完全动态的场景。

我一般推荐用 Put，因为可读性和可维护性更好，性能差距在实际业务中并不明显。下面是一个典型的转换函数：

import org.apache.hadoop.hbase.client.Put import org.apache.hadoop.hbase.io.ImmutableBytesWritable import org.apache.hadoop.hbase.util.Bytes def buildPut(rowKey: String, columns: Map[String, String]): (ImmutableBytesWritable, Put) = { val put = new Put(Bytes.toBytes(rowKey)) columns.foreach { case (col, value) => val parts = col.split(":", 2) if (parts.length == 2) { put.addColumn(Bytes.toBytes(parts(0)), Bytes.toBytes(parts(1)), Bytes.toBytes(value)) } } (new ImmutableBytesWritable(Bytes.toBytes(rowKey)), put) }

这段代码里，columns的 key 是family:qualifier格式，比如b:name、t:tag_001。这样设计的好处是，无论列族和列限定符怎么变，转换逻辑都不用改。

4.3 配置 HFileOutputFormat2

HFileOutputFormat2需要知道目标表的元信息，包括表名、列族、Region 分布等。这些信息通过Configuration传递：

import org.apache.hadoop.hbase.HBaseConfiguration import org.apache.hadoop.hbase.mapreduce.HFileOutputFormat2 import org.apache.hadoop.hbase.client.ConnectionFactory import org.apache.hadoop.hbase.TableName import org.apache.hadoop.mapreduce.Job val hbaseConf = HBaseConfiguration.create() val connection = ConnectionFactory.createConnection(hbaseConf) val tableName = TableName.valueOf("user_profile") val table = connection.getTable(tableName) val regionLocator = connection.getRegionLocator(tableName) val job = Job.getInstance(hbaseConf) HFileOutputFormat2.configureIncrementalLoad(job, table, regionLocator) val jobConf = job.getConfiguration

configureIncrementalLoad这个方法很关键，它会自动配置：

• 输出的 KeyValue 排序规则
• 分区器（按 Region 边界分区）
• 输出格式类

配置好之后，就可以用 Spark 写 HFile 了：

val rdd: RDD[(ImmutableBytesWritable, Put)] = ... // 转换后的 RDD rdd.saveAsNewAPIHadoopFile( hfileOutputPath, classOf[ImmutableBytesWritable], classOf[Put], classOf[HFileOutputFormat2], jobConf )

注意：saveAsNewAPIHadoopFile的第二个参数是 key 类型，第三个是 value 类型，第四个是 OutputFormat 类。这里 key 必须是ImmutableBytesWritable，value 是Put，不能搞错，否则会报类型转换异常。

4.4 分区数与 HFile 数量的关系

Spark RDD 的分区数直接影响生成的 HFile 数量。每个 Spark 分区会生成一个 HFile（或者多个，取决于是否有 spill）。如果分区数太多，会生成大量小 HFile，加载时元数据操作开销大；如果分区数太少，单个 HFile 太大，加载时移动文件慢。

我的经验是，分区数设置为 Region 数量的 1-2 倍比较合适。比如表有 10 个 Region，Spark 分区数设 10-20。这样每个 Region 对应 1-2 个 HFile，加载效率最高。

可以通过rdd.repartition(N)来调整分区数，但要注意 repartition 会触发 shuffle，如果数据量很大，shuffle 成本不低。更好的做法是在数据转换阶段就控制好并行度，比如从 Hive 读数据时用spark.sql.shuffle.partitions控制。

5. 加载 HFile 到 HBase

5.1 LoadIncrementalHFiles 的用法

HFile 生成好之后，用LoadIncrementalHFiles工具加载。这个工具既可以命令行调用，也可以在代码里调用。

命令行方式：

hbase org.apache.hadoop.hbase.mapreduce.LoadIncrementalHFiles /tmp/hfile_output user_profile

代码方式：

import org.apache.hadoop.hbase.mapreduce.LoadIncrementalHFiles val loader = new LoadIncrementalHFiles(hbaseConf) loader.doBulkLoad(new Path(hfileOutputPath), admin, table, regionLocator)

代码方式的好处是可以集成到 Spark 任务里，生成完 HFile 直接加载，不用人工介入。但要注意，doBulkLoad是阻塞的，加载完成前任务不会结束。

5.2 加载过程中的关键参数

LoadIncrementalHFiles有几个参数值得关注：

• hbase.bulkload.retries.number：加载失败时的重试次数，默认 10。如果集群不稳定，可以适当调大。
• hbase.bulkload.retries.retryOnClusterException：是否在集群异常时重试，默认 false。生产环境建议设为 true。
• hbase.loadincremental.threads.max：加载时的并发线程数，默认是 Region 数量。如果 Region 很多，可以适当调大，但不要超过集群承受能力。

这些参数可以在hbase-site.xml里配置，也可以在代码里通过Configuration.set()动态设置。

5.3 加载后的验证

加载完成后，一定要验证数据是否正确写入。最简单的方式是用count命令统计行数，或者用scan抽样查看。

count 'user_profile' scan 'user_profile', {LIMIT => 5}

如果行数和预期不符，或者某些列缺失，就要回头检查 HFile 生成阶段的问题。常见的验证点包括：

• rowkey 是否正确
• 列族和列限定符是否和表结构一致
• 列值是否被正确序列化（比如数字被转成了字符串）

我一般会写一个校验脚本，从 HBase 里抽样读一批数据，和源数据做比对，确保字段级的一致性。这个步骤在数据质量要求高的场景下不能省。

6. 常见问题与排查技巧实录

6.1 报错速查表

6.2 几个我踩过的坑

坑一：列族顺序问题。HBase 的 KeyValue 是按rowkey + family + qualifier + timestamp排序的。如果 HFile 里的 KeyValue 顺序不对，加载时会报错。用HFileOutputFormat2一般不会出这个问题，因为它内部会做排序。但如果你自己构造 KeyValue，就要特别注意。

坑二：时间戳冲突。如果同一行同一列有多个版本，BulkLoad 会全部加载进去。如果表设置的VERSIONS是 1，加载后只会保留最新版本。这个行为有时候会导致数据"丢失"的错觉，其实是版本被覆盖了。建议在生成 HFile 时就控制好版本，只保留需要的那一版。

坑三：临时目录和 HBase 根目录不在同一集群。LoadIncrementalHFiles是通过文件移动来加载的，如果临时目录和 HBase 根目录不在同一个 HDFS 集群，移动会变成跨集群拷贝，非常慢，甚至失败。一定要确保两者在同一集群。

坑四：Spark 任务的 executor 内存不足。生成 HFile 时，如果单行数据特别大（比如一行有几百个列），executor 内存容易被撑爆。这时候要调大spark.executor.memory，或者把大行拆分成多行。

6.3 性能调优的几个方向

• 控制 HFile 大小：单个 HFile 控制在 1-2GB 比较合适，太大加载慢，太小元数据多。
• 批量加载：如果数据量特别大，可以分批次生成和加载，避免一次性占用太多资源。
• 压缩算法：HFile 支持压缩，常用的有 Snappy、LZ4。压缩能显著减少 HDFS 空间和 IO，但会增加 CPU 开销。根据集群情况选择。
• 预分区 + 加盐：前面讲过，这是避免热点最有效的手段。

7. 一些实战经验与扩展思路

7.1 增量数据的处理

BulkLoad 适合大批量离线写入，但如果是增量数据（比如每小时一批），每次都全量生成 HFile 就不合适了。这时候可以按时间分区，每次只处理新增的数据，生成对应的 HFile 加载。

但要注意，增量加载时 rowkey 的设计要避免和已有数据冲突。如果 rowkey 是userId + timestamp，增量数据的时间戳是新的，不会冲突。如果 rowkey 只有userId，增量加载会覆盖已有数据，这时候要确认是否符合业务预期。

7.2 和 Spark SQL 的结合

如果源数据在 Hive 表里，可以用 Spark SQL 读取，然后转换成 RDD 再生成 HFile。这样能充分利用 Spark SQL 的优化能力，比如谓词下推、列裁剪。

val df = spark.sql("SELECT user_id, name, age, city FROM user_source") val rdd = df.rdd.map { row => val userId = row.getAs[String]("user_id") val columns = Map( "b:name" -> row.getAs[String]("name"), "b:age" -> row.getAs[Int]("age").toString, "b:city" -> row.getAs[String]("city") ) buildPut(userId, columns) }

这种方式代码简洁，而且 Spark SQL 会自动优化读取。但要注意，df.rdd会触发从 DataFrame 到 RDD 的转换，失去一些 Catalyst 优化，所以尽量在 SQL 阶段把能做的过滤和聚合都做完。

7.3 监控和告警

BulkLoad 任务一般跑在离线调度系统里，监控很重要。我通常会关注这几个指标：

• 任务耗时：突然变长可能是数据量增加或集群资源紧张。
• HFile 数量：异常增多可能是分区数设置不合理。
• 加载失败次数：频繁失败要检查集群状态。
• HDFS 空间：临时目录不及时清理会慢慢吃满空间。

这些指标可以接入现有的监控系统，设置阈值告警。别等到任务失败了才发现问题。

7.4 一个容易被忽视的细节：HFile 的清理

前面提过，BulkLoad 生成的 HFile 在加载完成后，理论上可以删除。但LoadIncrementalHFiles默认不会删除源文件，需要手动清理。如果不清理，HDFS 上会积累大量临时文件。

我的做法是在加载成功后，用一个独立的步骤删除临时目录：

val fs = FileSystem.get(hbaseConf) fs.delete(new Path(hfileOutputPath), true)

但要注意，如果加载失败需要重试，就不能提前删除。所以清理逻辑要放在确认加载成功之后。另外，如果多个任务共用同一个临时目录，清理时要小心不要误删其他任务的文件。最好每个任务用独立的目录，比如带上任务 ID 或时间戳。

这套方案我在几个不同规模的项目里都用过，从几百万行到几十亿行都跑得通。核心就是理解 BulkLoad 的原理，把 rowkey、列族、分区这几件事设计好，剩下的就是按部就班地实现和调优。真正难的不是代码，而是对业务数据分布的理解和对 HBase 存储机制的把握。