HBase Bulk Load Architecture in Depth

What bulk loading actually is

HBase's normal ingest path charges every cell the same tolls: an RPC from the client to a RegionServer, an append to the write-ahead log, a sync that a handler thread waits on, and space in the MemStore until a flush turns it into an HFile. Bulk loading skips all of it. A separate job writes finished HFiles into HDFS, and a second step hands those files to the RegionServers, which move them into the store directory of whichever region owns their key range.

The consequence is that ingest stops being limited by how fast RegionServers can absorb mutations and starts being limited by how fast your cluster can write and rename files in HDFS. That is why it is the standard answer for a backfill, a migration off another store, or a nightly rebuild of a derived table: the expensive part moves into a MapReduce or Spark job you already know how to scale, and the serving cluster does almost nothing until the final move.

The price is that everything the write path was quietly doing on your behalf stops happening. Replication enrolment, coprocessor hooks, sequencing against in-memory data and version semantics all change. Most of this page is about those consequences, because they are where bulk loads actually go wrong - the ingest part works on the first try.

Stage one: writing HFiles the table will accept

The first stage is an ordinary MapReduce or Spark job whose output format is HFileOutputFormat2. Rather than emitting records to a sink, it writes files in HBase's own on-disk format under a directory tree with one subdirectory per column family - <output>/d/<files>, <output>/meta/<files>. The load step later uses those directory names to decide which store each file belongs to, so a misspelt family name is not caught by the job at all; it surfaces as a load failure long after the compute has been paid for.

The files must be readable by the table exactly as it is configured today. HFileOutputFormat2 reads the target's table descriptor and inherits per-family compression, block size, data block encoding and bloom settings, which is the main argument for letting it configure itself instead of hand-setting those properties in your driver. Set a codec the RegionServers cannot load and the load itself succeeds - the files are just bytes to a rename - while every subsequent read of the affected regions fails. What the writer is actually laying down, the block index, the trailer, the encoding choices, belongs to the HFile format.

Two ordering rules are absolute, and the second one is the reason the next section exists. Cells within a single file must be in ascending key order, and the writer enforces it rather than trusting you: hand it a key that does not sort after its predecessor and the task dies on the spot instead of producing a file that would fail mysteriously at read time. And each file must lie entirely inside one region's key range.

configureIncrementalLoad and the total-order partitioner

HFileOutputFormat2.configureIncrementalLoad(job, table, regionLocator) is one call that rewrites most of the job's configuration, and understanding what it changes is most of understanding bulk load. It sets the output format; it asks the RegionLocator for the table's current region start keys; it sets the number of reduce tasks to the number of regions; it installs a total-order partitioner, writing the region boundaries into a partition file that is shipped to every task through the distributed cache; it selects a reducer that sorts by cell key, choosing between the Put, KeyValue and Cell sort reducers based on the map output value class you declared; and it copies the per-family storage settings out of the table descriptor.

The partitioner choice is the load-bearing part. A default hash partitioner scatters adjacent keys across every reducer, so every reducer would emit files spanning the whole key space, and every one of those files would straddle every region boundary. A total-order partitioner instead guarantees that reducer i receives exactly the keys that fall inside region i, so each reducer emits one file per family that fits one region precisely.

This makes the reducer count a derived value, not a tuning knob. Overriding mapreduce.job.reduces after calling configureIncrementalLoad breaks the correspondence between partitions and regions while leaving the job apparently healthy; you get output, the load accepts it, and the load quietly does far more work than it should because most files now need splitting.

Pre-splitting as partitioner alignment

Because the reducer count is the region count, the shape of the table before the job runs dictates the shape of the job. A table created with no split points has exactly one region, so the job gets exactly one reducer: the entire dataset funnels through a single task, sorts on one node, and produces one enormous HFile. It usually finishes eventually, and it is frequently slower than the API writes you were trying to avoid.

Pre-splitting fixes both ends at once. Create the table with explicit boundaries - SPLITS in the shell, or a region count plus a split algorithm - and the partitioner has real boundaries to work with, the reduce phase parallelises, and the loaded data arrives already distributed across RegionServers instead of on one.

Choosing the boundaries is a row key question, and key distribution is developed at length in hotspotting. What is specific to bulk load is the count and where the numbers come from. Too few regions gives too few reducers, files large enough that a mid-load split has to rewrite gigabytes, and a final move that serialises on a handful of servers. Too many gives thousands of small files landing on every region at the same instant, which is store-file pressure everywhere at once. And the boundaries should come from quantiles of a sample of the actual source keys, not a uniform range split: a total-order partitioner over skewed keys leaves one reducer running for hours after the others have finished, and no amount of cluster capacity helps.

Stage two: the load, and what atomic covers

The second stage is the load tool, usually invoked through its hbase completebulkload <outputdir> <table> shorthand. It walks the output directory, groups files by family, consults hbase:meta for the region owning each file's key range, and issues a bulk-load RPC to the RegionServer hosting that region. The server takes the region's lock, moves each file into the store directory, assigns it a sequence id so its cells order correctly against anything already sitting in the MemStore, and makes it visible to readers.

Atomic here means atomic per region, per family. All the files handed to one region in one call become visible together or not at all. It does not mean the load as a whole is atomic. A load touching five hundred regions is five hundred independent operations spread over as many seconds as they take, and a scanner running during the load can read a key range that has landed immediately next to one that has not. If a consumer needs all-or-nothing visibility over the whole dataset, that has to come from somewhere other than the load - loading into a staging table and swapping names, or gating readers for the duration.

The move is an HDFS rename when the job's output directory and the HBase root directory live on the same filesystem, which is why terabytes land in seconds. Point the output at a different filesystem and the identical command silently degrades into a byte-for-byte copy of everything, turning the cheapest stage of the pipeline into the most expensive one.

When a region splits between the write and the load

Nothing freezes the table between the two stages. The boundaries the partitioner used are a snapshot taken when the job was configured; by the time the load runs, the balancer may have moved regions, a write-heavy region may have split, and an operator may have merged two others.

The load copes rather than failing. When it finds a file whose key range crosses a current boundary, it splits that file in two at the boundary - writing a top and a bottom half - and retries placement for each half independently, recursing if a half still does not fit. The recursion is bounded by hbase.bulkload.retries.number, ten by default; exhaust it and the tool gives up with some files placed and some not, which is the messiest state bulk load has.

The path is correct but it is precisely the cost bulk load exists to avoid: splitting an HFile means reading it and rewriting it, on the client, single-threaded. If hours pass between the job and the load, or the table is actively splitting under other traffic, you pay it on most of your files and the load stops looking fast. Keeping the two stages adjacent, and pausing the balancer or automatic splits for the duration of a large load, is what keeps the rename path the one that actually runs.

Permissions, ownership, and the staging directory

The files are written by the job's user and consumed by the RegionServer running as the HBase service user. On a permissive development cluster nobody ever notices; on a Kerberised cluster with a restrictive umask this is the single most common way a first bulk load fails, and the error message points at a user who never touched the file.

The mechanism that reconciles the two identities is a staging directory, hbase.bulkload.staging.dir, conventionally under the HBase root and owned by the HBase user. The output is staged there first, and the server performs the final move into the store directory with its own credentials. Two requirements fall out of that: the job user must be able to create and write inside the staging area, and the HBase user must be able to read what lands there. A restrictive umask on the job side is the usual reason the second one is not met, and it is worth checking before blaming Kerberos.

The subtler failure is ownership rather than permission. Files that HBase can read but does not own sit inside a store directory it manages under an identity it did not create, and the consequences surface later, during whatever next rewrites or relocates those files, rather than at load time - which makes them hard to connect back to the load that caused them. Running the load step as the HBase user, or making the staging handover do the ownership change, avoids the whole class.

Replication does not see it

By default a bulk-loaded file is invisible to replication. Nothing was appended to the WAL, and tailing the WAL is the entire mechanism, so the peer stays perfectly healthy, reports no lag, and is missing the data. This is a silent divergence that surfaces at failover, which is the worst possible moment to discover it. Enrolling bulk loads is a source-side configuration change with real cross-cluster requirements, and it is covered in replication. From the loader's side the rule is simpler: after any bulk load into a replicated table, verify against the sink directly. Peer metrics cannot report a gap they were never told about.

The coprocessor hooks that never fire

Bulk load bypasses the write path, so it bypasses every observer hook attached to the write path. prePut and postPut do not run. Neither does anything hanging off the batch mutation hooks. There are dedicated bulk-load hooks - the pre and post variants around loading an HFile - but they receive a file and a path, once per file, not a cell and a value once per row. They are the right place for an audit record or an authorisation check and the wrong place for anything that has to look at the data, because looking at the data means opening and scanning the file yourself.

The practical consequence is worth stating plainly, because it costs people real correctness: any secondary index maintained by a coprocessor is not updated by a bulk load. The base table gains a hundred million rows and the index gains nothing, and every index-driven query afterwards returns a result that looks correct and is incomplete. SQL layers built over HBase handle this explicitly, with their own bulk loader that builds the index tables in the same job, precisely because retrofitting the index afterwards means a full rescan of what you just loaded. If you maintain the index yourself, the index has to be a second bulk load derived from the same source data in the same pipeline - not a follow-up job that reads the table back.

Compaction pressure and locality, after the load reports success

A bulk load adds one file per family to every region it touches, simultaneously. On a table whose compaction threshold is a handful of files, one load can push every region over the trigger at the same moment, and the cluster spends the next several hours compacting while read latency looks nothing like it did the day before. Frequent small loads are the pathological version: each adds another file everywhere, and eventually the blocking store file limit, hbase.hstore.blockingStoreFiles, stalls writes to a region outright. Policy and tuning belong to compaction; what belongs here is scheduling - batch so each load carries enough data to justify the file it leaves behind, and consider hbase.mapreduce.hfileoutputformat.compaction.exclude when you intend to run a major compaction yourself afterwards and do not want minor compactions churning the same bytes first.

Locality is the second aftershock and the less obvious one. The reducers wrote their output wherever the tasks happened to be scheduled, so whether the RegionServer that ends up owning a file has a local replica of its blocks is incidental rather than guaranteed, and in practice it is often poor. Reads against freshly loaded regions therefore cross the network far more often than reads of data that was flushed locally, and the remedy is a major compaction, which rewrites the region's files through the local DataNode and restores locality as a side effect. On a large one-time load this is not optional tidying afterwards; budget it as the last stage of the load.

A load, end to end

The shell side is short. The interesting decisions all happened before it.

# 1. Create the table pre-split on boundaries sampled from the source keys.
create 'orders', {NAME => 'd', COMPRESSION => 'SNAPPY'}, \
  {SPLITS => ['1000','2000','3000','4000']}

# 2. Generate HFiles: writes /tmp/bulk/orders/d/<one file per region>.
#    Do NOT override the reducer count the driver derived from the table.
hadoop jar my-etl.jar com.example.OrdersBulkLoad \
  /warehouse/orders/2026-08-01 /tmp/bulk/orders

# 3. Move them in. Seconds for terabytes, on one filesystem.
hbase completebulkload /tmp/bulk/orders orders

The driver is where the configuration actually happens. Declaring the map output value class before the configure call is what selects the sort reducer, and both the Table and the RegionLocator are needed - the first for the descriptor, the second for the boundaries.

Job job = Job.getInstance(conf, "orders-bulk-load");
job.setMapperClass(OrdersMapper.class);
job.setMapOutputKeyClass(ImmutableBytesWritable.class);
job.setMapOutputValueClass(Put.class);        // selects the Put sort reducer

TableName name = TableName.valueOf("orders");
try (Connection conn = ConnectionFactory.createConnection(conf);
     Table table = conn.getTable(name);
     RegionLocator locator = conn.getRegionLocator(name)) {
    // sets output format, reducer, reducer count, partitioner,
    // partition file, and per-family compression / block size / encoding
    HFileOutputFormat2.configureIncrementalLoad(job, table, locator);
}
FileOutputFormat.setOutputPath(job, new Path("/tmp/bulk/orders"));
job.waitForCompletion(true);

In Spark the same output format is used directly: sort by the full cell key, partition on the region boundaries yourself, and write with saveAsNewAPIHadoopFile against a job that configureIncrementalLoad has already configured. The reducer selection is irrelevant there because there is no reducer - which means the ordering and partitioning guarantees the MapReduce path gave you for free are now your responsibility, and getting the partitioning wrong shows up as a load that spends its time bisecting files rather than as an error.

Re-running a load, and what idempotency rests on

The load moves files out of the output directory, so running completebulkload against the same path twice does nothing the second time: the directory is empty. That is a useful property when a load is interrupted, and it is not the question people usually mean.

Re-running the whole pipeline is the real question, and the answer is decided entirely by timestamps. If the job sets an explicit timestamp on every cell, derived from the source record rather than from the clock, then a second run produces cells identical in row, family, qualifier and timestamp to the first, and HBase keeps one of them - the load is genuinely idempotent. If the job lets the timestamp default to write time, the second run produces a distinct version of every single cell. Under VERSIONS => 1 the older copy is shadowed on read and lives on disk until a compaction drops it, so a re-run silently doubles the store footprint for a while. Under a higher version count both versions are real, and a scan requesting all versions returns both.

So bulk load is exactly as idempotent as your timestamp discipline, and that is a decision worth making before the first production run rather than in the middle of recovering from a failed one.

When bulk load is the wrong choice

It is wrong whenever the fixed overhead dominates. A few thousand rows do not justify a cluster job, a region lookup and a file per region; batched Puts through the ordinary client are faster end to end and leave nothing behind for compaction to clean up. Streaming ingest is the same argument at a higher frequency - a load every minute is a new file on every region every minute, and the store file count wins that race.

It is wrong when the write needs to read. checkAndPut, Increment and Append are evaluated against current state on the RegionServer; there is no bulk equivalent, because the file was finished before HBase ever saw it. Any conditional or accumulating semantics have to be resolved in the job, against a snapshot of state you fetched yourself, with all the staleness that implies.

And it is wrong when something downstream depends on the write path. The replication and coprocessor gaps above are not defects to be tuned around; they follow directly from the design, and if you cannot accept them the honest answer is to write through the API and pay the cost. Finally, if the goal is to move a table that already exists rather than build one from source data, this is not the tool: snapshots copy file references rather than regenerating files, and do it without a compute job at all.