Hive数据导入HBase性能优化：BulkLoad原理与Spark/MapReduce实现对比

之前接了个需求，要把Hive里的用户行为标签表每天灌到HBase里给实时查询用。第一版我图省事，直接用Java API写了个Put任务，一条一条塞。结果5亿条数据任务跑了6个小时没跑完，还把RegionServer的CPU打满了，线上业务直接受到牵连。后来我改成BulkLoad方式，同样一批数据，MapReduce版本十几分钟完成，Spark版本更快，而且对在线集群几乎没有影响。这篇就以这个场景为线索，讲清楚Hive数据导入HBase使用BulkLoad这事，重点拆Spark和MapReduce两种实现各自的本质、代码套路和那些文档里不会写的坑。

如果你刚接触HBase，或者正在被海量数据写入性能问题折磨，这篇文章值得你花时间看完。它不光是给一份能跑的代码，还会解释清楚为什么BulkLoad能快这么多、两种实现该怎么选、以及跑了以后数据出问题该怎么排查。

1. BulkLoad快得多的根源：先打包好，再直接上架

很多教程一上来就贴代码，可代码跑不通的时候你根本不知道怎么调。所以我先讲原理，搞懂原理之后，代码怎么写、参数怎么调，你心里自然就有数了。

1.1 常规写入链路里最重的两个开销

如果你用HBase自带的API做Put，每一条记录从客户端到真正落盘，要经过这么一条链路：客户端构造Put请求，经过RPC发给RegionServer，RegionServer先写WAL（预写日志），再写入MemStore，MemStore攒到一定阈值后flush成StoreFile，StoreFile多了还要做Compaction，Region太大了还要Split。

这里面最重的两个开销，第一是WAL的写入。每条Put都得先追加一条日志，为了数据安全通常还要把持久化级别设置为同步刷盘，这本身就是一次磁盘IO。第二是RPC的网络开销。客户端和RegionServer之间一次请求只能带有限数量的Put，数据量大的时候，网络包往返次数是惊人的。这两者叠加，就是常规写入在大批量场景下性能上不去的根本原因。

1.2 HFile的生成与加载，绕开了什么

BulkLoad的思路完全不一样。它先把数据在离线计算引擎里直接构造成HBase底层的存储文件HFile，然后把HFile文件整体放到HDFS上指定的Region目录下，完成一次原子性的加载过程。

这个过程中间绕开了三件事：RPC链路、WAL日志、以及MemStore的flush过程。数据以文件形式直接落到Region的数据目录，HBase元数据更新一下就算写完了。你可以把常规Put想象成找搬家工人，一件一件从旧家搬到新家，每件都要填单、签字、过路；而BulkLoad是你自己雇人在旧家把所有东西按新家的房间分好类、打好包，然后一集装箱拉过去，到地方直接摆放就行。

这里要补充一句：BulkLoad没有经过WAL，所以如果RegionServer在加载过程中宕机了，这批数据不会通过日志重放来恢复。好在HFile文件本身已经持久化在HDFS上，HBase在Regoin上线时会主动扫描数据目录里的文件，所以实际丢失风险很小，但你不能按WAL那套"写完即肯定成功"的语义来理解它。

1.3 有序性：BulkLoad质量的分水岭

HBase的一张表按RowKey范围划分成多个Region，每个Region内部的StoreFile要求按照RowKey有序排列。BulkLoad生成HFile的时候，如果能按目标表在每个Region的边界范围内分别排序，加载时就能做到零Split、零Compaction，性能最好。

如果生成的HFile是全局乱序的，加载时HBase被迫对文件进行Split处理，把乱序的记录重新分配到对应Region，这个过程不仅慢，还可能引发大量额外的磁盘IO和IO忙。所以我一直跟团队强调：BulkLoad性能好不好，不看你代码写得多花哨，而看你有没有按Region边界做好数据分区和排序。

这也是为什么标题里两种实现的核心思路其实是一致的，只是计算框架不同。下一节先讲共同的前置条件，再分别讲代码。

2. 动手前的准备工作：表设计、依赖和环境校验

BulkLoad不是把数据文件往HDFS上一扔就完事。它依赖你的HBase表结构、Region分布、以及Hive侧的数据形态。这些没准备到位，后面跑起来全是怪毛病。

2.1 典型的版本组合和依赖坐标

以我实际用的社区版组合为例：Hadoop 2.10、Hive 2.3、HBase 1.4/2.x、Spark 2.4。这套组合比较常见，网上资料也多。不同大版本的API签名会有一点差异，但核心类没有变过：HFileOutputFormat2、LoadIncrementalHFiles、ImmutableBytesWritable。

如果你用Maven，核心依赖是这些：

<dependency> <groupId>org.apache.hbase</groupId> <artifactId>hbase-client</artifactId> <version>2.1.0</version> </dependency> <dependency> <groupId>org.apache.hbase</groupId> <artifactId>hbase-server</artifactId> <version>2.1.0</version> </dependency> <dependency> <groupId>org.apache.hbase</groupId> <artifactId>hbase-mapreduce</artifactId> <version>2.1.0</version> </dependency>

注意一点：hbase-server这个依赖会把一系列HBase内部类带进classpath，很容易和Hadoop集群自带的jar包冲突。我在本地调试时经常遇到NoClassDefFoundError，排查半天发现是jar包顺序问题。解决办法是使用mvn dependency:tree看清楚传递依赖，把冲突的排除掉，或者直接用shade插件打包。

2.2 RowKey设计和列簇数量怎么定

BulkLoad对RowKey非常敏感，因为RowKey直接决定了数据在整个集群上的分布方式。如果Hive表里有用户id、日期这类字段，最常见的做法是把reverse(user_id) + 日期拼成RowKey，这样既保证同一用户的数据尽量连续，又能在查询时按用户维度快速定位。

关于列簇，我的建议是能用一个尽量用一个。列簇越多，意味着一个Region里要维护的Store就越多，BulkLoad时生成的HFile个数也会成倍增加，加载时RegionServer的压力呈线性上涨。除非你确实有"冷热数据不同压缩策略"这种强需求，否则单列簇是默认选择。

2.3 检查Hive侧数据：路径、分区、空值、类型

正式写代码前，先花十分钟看一眼Hive表在HDFS上的实际文件。你可以用hadoop fs -ls检查表目录，也可以在Hive里直接desc formatted table_name查看location。这一步有两个目的：一是确认数据文件确实存在，二是看文件数量和数据量级。

文件数量决定了你该开多少个Reduce分区。如果你有200个小文件，却只开10个Reduce，每个Reduce要读20个文件，Mapper的启动开销和网络读取压力会非常明显。反过来如果只有一个大文件却开1000个Reduce，又会生成大量小HFile，后面还是得靠Major Compaction合并。

空值和类型转换也要提前处理。Hive里的NULL在HBase的Put里不能直接放进去，它会被当成空字节数组，可能造成你查询时数据对不上。我会在转换前统一做一次过滤或者用默认值替换。类型方面，Hive的BIGINT和HBase字节转换要注意大小端问题，我习惯统一走Bytes.toBytes，避免手写转换时踩符号位扩展的坑。

3. MapReduce实现：用HBase自带的输出格式稳稳跑通

MapReduce是最传统的实现方式，也是理解BulkLoad机制的最佳切入路径。它代码量大一些，但逻辑直接：读Hive文件，按表Region边界分区排序，产出HFile，再加载。

3.1 为什么选HFileOutputFormat2而不是自定义OutputFormat

HBase自带了一个HFileOutputFormat2，它专门负责把MapReduce的输出写成本地文件系统能识别的HFile格式。你当然可以自定义OutputFormat自己写文件，但要处理KV编码、数据块压缩、布隆过滤器、魔数头等一堆细节，完全没有必要。

更重要的是，HFileOutputFormat2配合configureIncrementalLoad方法会自动根据目标表的Region边界，为每个Reduce分配一个RowKey区间。它内部用的是TotalOrderPartitioner，也就是说Map端输出的数据经过Combine后，会按照你HBase表当前的Region起始Key做全局排序。这一步是BulkLoad质量的关键，你自己手写分区逻辑很难做对。

3.2 Mapper里要把Hive行翻译成什么

Mapper的输入是Hive表在HDFS上的原始文件行，输出结果是ImmutableBytesWritable和Put。第一个泛型用来做分区排序的Key，第二个泛型是实际要写入的数据。直接看代码：

public class HiveToHBaseMapper extends Mapper<LongWritable, Text, ImmutableBytesWritable, Put> { @Override protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException { String[] fields = value.toString().split("\u0001"); if (fields.length < 4) { return; } // 假设Hive字段顺序：user_id, event_time, event_type, amount String rowKey = reverse(fields[0]) + "_" + fields[1].replace("-", ""); Put put = new Put(Bytes.toBytes(rowKey)); // 根据Version值让最新数据覆盖旧数据，这里也可以用固定的时间戳 long ts = Long.parseLong(fields[1].replace("-", "").replace(":", "").replace(" ", "")); put.addColumn(Bytes.toBytes("cf"), Bytes.toBytes("event_type"), ts, Bytes.toBytes(fields[2])); put.addColumn(Bytes.toBytes("cf"), Bytes.toBytes("amount"), ts, Bytes.toBytes(fields[3])); context.write(new ImmutableBytesWritable(Bytes.toBytes(rowKey)), put); } private static String reverse(String s) { return new StringBuilder(s).reverse().toString(); } }

这里我做了两个关键决定：第一，RowKey把用户id做了反序，能打散热点用户造成的访问倾斜；第二，时间戳直接用事件发生时间而不是默认的当前时间，这样HBase里同一RowKey下不同版本的数据能按业务时间排序。这个设计在后面对账时帮了我大忙。

3.3 configureIncrementalLoad和Reduce阶段到底做了什么

核心配置在Driver类里完成。假设你已经根据Hive的分区条件筛选好了数据文件，下面这段是主体逻辑：

Configuration conf = HBaseConfiguration.create(); conf.set("hbase.zookeeper.quorum", "zk1,zk2,zk3"); Job job = Job.getInstance(conf, "HiveDataBulkLoad"); job.setJarByClass(BulkLoadDriver.class); // 输入：Hive表所在HDFS路径 FileInputFormat.addInputPath(job, new Path("/user/hive/warehouse/ods_user_events/dt=2025-01-01")); TableName tableName = TableName.valueOf("user_events"); Connection conn = ConnectionFactory.createConnection(conf); Table table = conn.getTable(tableName); RegionLocator locator = conn.getRegionLocator(tableName); // 关键：按表Region边界自动设置分区 HFileOutputFormat2.configureIncrementalLoad(job, table.getDescriptor(), locator); job.setMapperClass(HiveToHBaseMapper.class); job.setMapOutputKeyClass(ImmutableBytesWritable.class); job.setMapOutputValueClass(Put.class); job.setOutputFormatClass(HFileOutputFormat2.class); FileOutputFormat.setOutputPath(job, new Path("/tmp/bulkload_output")); boolean ok = job.waitForCompletion(true); if (ok) { LoadIncrementalHFiles loader = new LoadIncrementalHFiles(conf); loader.doBulkLoad(new Path("/tmp/bulkload_output"), conn.getAdmin(), table, locator); }

configureIncrementalLoad这一步是关键中的关键，它扫描了目标表当前所有Region的起始RowKey，转换成一堆分区边界写入到job的配置里。Reduce阶段开始前，Map端输出会经过一次全排序，同一个Reduce拿到的是RowKey完全连续的一段数据，它再把这些数据顺序写入HFile，这样就保证了StoreFile的有序性。

Reduce跑完会在输出目录生成一批后缀为.hfile的文件。此时数据还没进HBase，只是躺在HDFS临时目录里。LoadIncrementalHFiles会逐个文件判断它落在哪个Region范围内，通过RegionServer做一次文件级别的批量加载。这个过程快得惊人，因为本质上就是移动目录或硬链接，很少涉及数据拷贝。

3.4 这个方案最容易踩的三个配置坑

第一个坑是运行时classpath没有包含HBase依赖。你本地跑通了，放到集群上就报ClassNotFoundException: org.apache.hadoop.hbase.mapreduce.HFileOutputFormat2。十有八九是job的libjars没传或者打包时用了provided scope。我的习惯是直接把hbase-mapreduce的jar包用-libjars参数带上去。

第二个坑是fs.defaultFS两端不一致。HDFS上的输入路径用的是默认文件系统，临时目录输出却写到本地文件系统上，最后加载时HBase尝试读一个本地不存在或权限不对的路径，莫名其妙的空指针就这么来的。建议代码里对所有路径统一使用全路径写法，比如hdfs://nameservice/user/...。

第三个坑是时区导致的RowKey拼接错误。Hive里的日期是字符串还好说，如果是TIMESTAMP类型且转成Date对象，在代码里格式化时默认时区是GMT，而你集群时区是Asia/Shanghai。两者相差8小时，最终RowKey里日期段就会少一天。这类问题不会报错，只会在你查询时发现该有的数据总是缺一个边界。排查方式也很简单，加日志打印一下格式化前后的值。

4. Spark实现：从DataFrame到HFile再到在线表

MapReduce方案没问题，但日常开发里大家更愿意用Spark，因为代码更简洁、数据处理能力强，还能复用一套已经写好的ETL逻辑。Spark做BulkLoad的核心思路和MapReduce完全一致，只是把MapReduce的两个阶段换成了RDD算子。

4.1 Spark读Hive的路线选择

读Hive有三种常见做法，实际效果差别很大。

第一种是spark.sql("SELECT * FROM ods_user_events WHERE dt='2025-01-01'")，走Spark SQL的Catalyst优化器，能享受谓词下推和列裁剪，代码最友好。第二种是spark.table("ods_user_events")直接拿表，效果和第一种基本一样。第三种是绕过Hive元数据，直接spark.read.text("hdfs路径")读文件，自己解析分隔符，适合那种Hive表结构经常变、你不想被元数据束缚的场景。

大多数时候我推荐第一种。它的优势不只是性能，更在于类型是安全的，DataFrame里每个字段的类型都已经从Hive元数据里解析好了，不会出现我前面说的字符串日期和Timestamp混用导致的时区问题。

4.2 核心代码：转换、分区、写HFile、加载

先看完整流程，再逐个说明。下面的代码以Spark 2.4和HBase 2.1为例：

import org.apache.hadoop.conf.Configuration import org.apache.hadoop.fs.Path import org.apache.hadoop.hbase.{HBaseConfiguration, TableName} import org.apache.hadoop.hbase.client.{ConnectionFactory, Put, RegionLocator, Table} import org.apache.hadoop.hbase.io.ImmutableBytesWritable import org.apache.hadoop.hbase.mapreduce.HFileOutputFormat2 import org.apache.hadoop.hbase.mapreduce.LoadIncrementalHFiles import org.apache.hadoop.hbase.util.Bytes import org.apache.spark.sql.SparkSession object SparkHiveToHBaseBulkLoad { def main(args: Array[String]): Unit = { val spark = SparkSession.builder() .appName("HiveToHBaseBulkLoad") .enableHiveSupport() .getOrCreate() val hbaseConf: Configuration = HBaseConfiguration.create() hbaseConf.set("hbase.zookeeper.quorum", "zk1,zk2,zk3") val tempPath = "/tmp/hbase_bulk_load_output_" + System.currentTimeMillis() val tableName = TableName.valueOf("user_events") // 1. 从Hive读数据 val df = spark.sql("SELECT user_id, event_time, event_type, amount FROM ods_user_events WHERE dt='2025-01-01'") // 2. 转成 (ImmutableBytesWritable, Put) val hfileRDD = df.rdd.map { row => val userId = row.getAs[String]("user_id") val eventTime = row.getAs[String]("event_time").replace("-", "").replace(":", "").replace(" ", "") val rowKey = userId.reverse + "_" + eventTime val put = new Put(Bytes.toBytes(rowKey)) put.addColumn(Bytes.toBytes("cf"), Bytes.toBytes("event_type"), Bytes.toBytes(row.getAs[String]("event_type"))) put.addColumn(Bytes.toBytes("cf"), Bytes.toBytes("amount"), Bytes.toBytes(row.getAs[String]("amount"))) (new ImmutableBytesWritable(Bytes.toBytes(rowKey)), put) } // 3. 连接HBase，获取表描述和region信息 val conn = ConnectionFactory.createConnection(hbaseConf) val table: Table = conn.getTable(tableName) val locator: RegionLocator = conn.getRegionLocator(tableName) val job = org.apache.hadoop.mapreduce.Job.getInstance(hbaseConf) job.setMapOutputKeyClass(classOf[ImmutableBytesWritable]) job.setMapOutputValueClass(classOf[Put]) HFileOutputFormat2.configureIncrementalLoad(job, table.getDescriptor, locator) // 4. repartition并按HFileOutputFormat写出 hfileRDD.repartition(200) .saveAsNewAPIHadoopFile( tempPath, classOf[ImmutableBytesWritable], classOf[Put], classOf[HFileOutputFormat2], job.getConfiguration ) // 5. 加载HFile到HBase表 val loader = new LoadIncrementalHFiles(hbaseConf) loader.doBulkLoad(new Path(tempPath), conn.getAdmin, table, locator) conn.close() spark.stop() } }

这里的第4步有个细节值得注意：saveAsNewAPIHadoopFile会调用HFileOutputFormat2的getRecordWriter，在分布式环境下每个分区写一个HFile片段，配合前面configureIncrementalLoad设置好的分区器，Spark的shuffle会把这个RDD按RowKey全局排序后再写出。所以你看到的200个repartition，并不只是一个简单的扩大分区操作，它背后已经包含了sort阶段。

4.3 Spark方案怎么控制并行度和内存

Spark方案最容易出的问题就是Driver端OOM，因为saveAsNewAPIHadoopFile发生前，shuffle的sort数据要么落盘要么攒在内存里。实测下来，控制两个参数基本能避免OOM：spark.sql.shuffle.partitions和spark.executor.memory。

我的习惯是先把spark.sql.shuffle.partitions设成目标HFile数的2到3倍。目标HFile数不能拍脑袋定，可以用数据量除以单个HFile理想大小（128MB到256MB之间）来估算。5亿条记录大约50GB原始数据，除以128MB就是400个分区左右，repartition成600到800个是完全合理的。太多会产生小文件，太少则会延长shuffle时间。

内存方面，如果每个Executor内存是8GB，我会把spark.executor.memoryOverhead调到2GB以上，给shuffle sort留足堆外空间。另外spark.shuffle.file.buffer默认32KB也可以适当提高到256KB，减少磁盘IO次数。这些参数不一定要抄我的值，但思路是明确的：先估算数据量级，再定分区数，最后用内存参数兜底。

4.4 数据倾斜时的处理思路

Hive导入HBase最怕RowKey前缀分布极端。比如你的用户表里某个头部用户的记录占了40%，Spark的sort分区会严重不均，那个处理大分区的Executor直接成为瓶颈。

处理方式之一是加盐。在RowKey前面拼上一段2到3位的随机前缀，让数据的分布尽量均匀。但这样做有个代价，查询时你不知道原始RowKey的前缀是什么，无法直接依赖Region边界定位，必须全表扫描或者额外维护一个映射关系。所以我不建议盲目加盐。

更好的办法是看倾斜的粒度。如果是少量超大Key导致的倾斜，可以先把这些Key单独拎出来，用rdd.partitionBy自定义分区函数，按业务维度把超大Key拆散到多个分区，其他Key走默认分区。还有一种思路是改变RowKey本身的拼接顺序，比如把低基数的日期字段放前面，高基数的用户id放后面，这样即使某个用户数据多，也会被日期天然分隔开。这个调整在数据写入前做，代价最小。

5. 两种方案的实测对比与经验

纸上谈兵没有意义。我拿实际生产环境的数据做了几轮对比测试，只测同一份Hive数据往同一个HBase表里导入，硬件和参数尽量保持一致。数据量分别取1000万、5000万、1亿条三档。

5.1 两个方案的差异点对比

结果用一个表格说明。

别被这个表误导，MapReduce慢不代表它差。MapReduce方案的优势在于稳定：它不需要在内存里维持一个巨大的shuffle缓冲区，只要YARN队列有资源，几十亿数据也能老老实实跑完。我那个6小时跑不完的Java API版本，最后就是用MapReduce BulkLoad救回来的。

Spark优势是快，但它的快建立在Executor堆内存够用这个前提下。如果集群内存本来就紧张，频繁的Full GC会让Spark任务变慢，甚至不如MapReduce稳定。所以选哪种，本质是看你集群资源的情况和团队熟悉哪种框架。

5.2 我踩过的真实坑和排查链路

说实话，BulkLoad的代码只要跑通一次就不难，难的是出问题时你不知道该查哪层。这几个坑我每个都排查了很久，列出来供你少走弯路。

第一个坑是数据加载完查不到。症状是doBulkLoad返回成功，HBase表里文件也变多了，但get就是查不到。排查链路：先看RowKey拼接逻辑，再看列簇名和列名的大小写。HBase的列簇名是大小写敏感的，如果你建表时写的小写cf，代码里写的CF，文件照样生成了，但查询时HBase根本不会匹配到这个列。这类问题最难查，因为它不报错。

第二个坑是版本冲突。Hive表里同一主键有多条记录，BulkLoad时每个版本的时间戳相同，结果加载后只保留了一条。原因是你Put里没有设置时间戳，而多个Put在shuffle后落到同一个HFile的同一RowKey上时，HBase会按时间戳+Scanner顺序去掉重复版本。解决办法是在构造Put时为每条记录显式指定业务时间戳。

第三个坑是加载期间在线查询变慢。虽然BulkLoad比Put写入轻量，但大批量HFile一次性加载会让RegionServer的IO瞬间飙高，因为它要扫描目录、更新元数据，可能还会触发MemStore或StoreFile的合并调度。我的解决办法是把大任务拆成小批次，凌晨执行，或确保RegionServer的IO调度和查询任务的优先级错峰。

第四个坑是Region数据分布不均。前提是你建表时没有预分区，或者BulkLoad只是简单把所有文件往表里塞。排查链路：先hbase hbck或者用Web UI看Region大小分布；如果确认分布不均，回看configureIncrementalLoad的时候Region边界是否已经合理；最后决定是否重新建表并提前预分区。预分区数量和数据量比例可以按每个Region 10GB到20GB预估，这个值按你的单节点磁盘吞吐能力酌情调整。

5.3 增量导入的推荐做法

如果你每天只导当天新增的数据，没必要每次全量覆盖。增量导入的推荐方案是把导入逻辑设计成"目标时间范围+Offset"的模式：Hive侧用分区字段过滤当天的数据，HBase侧在加载前先根据RowKey前缀删除或跳过老数据。

实际实现时，我最常用的是两个小技巧。一是把日期字段直接拼进RowKey，这样每天的增量天然落在不同的RowKey区间，加载新数据不会和老数据冲突；二是加载前用Table.getScanner快速统计一下目标日期是否有数据，有的话先清理掉，避免出现同一天的数据被导了两次、版本号还互相覆盖的问题。

6. 长期运行视角：加载后的数据维护与巡检

BulkLoad让"Hive到HBase"这条路跑通了，但它不是终点。HFile文件加载完后，长期运行中还有很多细节关系到集群的稳定性。

6.1 加载完不等于万事大吉

BulkLoad直接生成并部署了StoreFile，它的文件大小和数量不会自动进入RegionServer的容量统计，至少不是即时生效。我遇到过一个现象：做一次大导之后，HBase的UI上Region大小看起来没变化，但实际磁盘占用已经上升了，过了很久才会在Major Compaction后修正。所以运维监控不能只盯HBase UI，要同时盯HDFS上Region目录的磁盘占用。

还有个相关的点：BulkLoad产出的HFile不会自动合并到已经存在的StoreFile里，它会在Region目录下作为独立文件存在。如果你的增量任务每天一次，一周后这个Region下面可能有7个甚至更多的StoreFile。StoreFile数量过多时，读取性能会下降，因为一次get可能要打开多个文件。所以增量导入后，要有意识地安排低峰期做Major Compaction。

6.2 Region平衡与合并策略

如果你用了预分区，Region基本能保持均匀。问题是数据写一段时间后，某些Region还是会因为RowKey分布变化而变大变小。这时HBase自带的Balancer能起一定作用，但它做的是Region数量层面的均衡，不是大小层面的均衡。

我在实践中发现，BulkLoad场景下的Region平衡最有效的手段还是源头控制：保证每次生成HFile时分区边界和表Region边界对得上，让数据从一开始就均匀散落。如果已经出现不均匀，你可以用split和move手工干预，但千万不要当作常态化操作。手工干预多了就说明RowKey设计有问题，得从源头调整。

6.3 我希望早点知道的一个小技巧

最后分享一个小技巧，是我踩过好几次坑之后总结出来的：给所有BulkLoad任务的输出路径加上数据日期和随机后缀，并且在任务开始前做一次路径存在性检查。

val tempPath = s"/tmp/hbase_bulk_load_${dt}_${System.currentTimeMillis()}"

这行代码看起来简单，实际价值很大。因为BulkLoad任务的临时目录一旦被前一个失败任务占用了，新任务写文件时会往同一个目录下继续堆，造成新老数据混在一起。而且旧的HFile文件如果不及时清理，会一直占用HDFS空间。我后来养成了一个习惯：每次任务结束、加载成功且验证数据无误后，立刻删临时目录。省下的可不止是磁盘空间，还有排查数据错乱时流的眼泪。

BulkLoad给我最大的体会是：大数据场景下，写入性能往往不在写入本身，而在你如何组织数据。只要理解了Region边界、StoreFile有序、离线生成在线加载这一条主线，Spark和MapReduce只是两个不同的工具而已。你完全可以把这里的代码改成Flink或者预先排好序的文件导入，思路都是同一套。希望这篇能把你的导入效率提上来，少踩几个我踩过的坑。