Impala + Kudu — Fast Analytical Updates on the Same Storage

The gap Kudu was built to fill

HDFS gives you sequential throughput and immutability, and the two are the same property. A file, once closed, cannot be modified; the only way to change a row is to rewrite the file or the whole partition containing it. For append-only event data that is a feature — immutability is precisely why Parquet scans are as fast as they are — but it is fatal for anything that gets corrected, backfilled, restated, or arrives late.

HBase covers the opposite corner. Single-row reads and writes by key in single-digit milliseconds, at effectively arbitrary write rates, with a sparse schema-flexible row model. What HBase does not do well is the full-table analytical scan: its on-disk layout is row-oriented and key-ordered, so a query touching three columns out of eighty still pays to read all eighty, and there is no columnar encoding to make the three cheap.

The workaround the industry converged on was to run both. Recent, mutable data landed in HBase; an hourly or nightly job converted settled data into Parquet on HDFS; analysts queried a view that unioned the two, taking care not to double-count the overlap window. Every part of that is real engineering cost — a second system to operate, a conversion job to babysit, a boundary condition to get right, and a freshness cliff exactly at the seam where the interesting data lives.

Kudu was designed to occupy the middle rather than bridge it. It stores data in columns, so scans read only what a query projects. It maintains a primary key index, so a single row can be located and rewritten without disturbing its neighbours. And it runs as its own service, with its own memory, write-ahead log and replication, instead of sitting on top of a filesystem that forbids mutation.

The storage model — a typed schema behind a mandatory primary key

Kudu is a structured storage engine: not a filesystem, not a key-value store, and not a query engine. Every table has a fixed, strongly typed schema declared up front, and every table has a primary key. There is no such thing as a Kudu table without one, and that key is the only index the table will ever have.

The rules around it are strict, and they are what most often forces a schema redesign on the way in:

• primary key columns must be listed first in the schema;
• every primary key column must be NOT NULL;
• a key column cannot be FLOAT, DOUBLE or BOOL;
• the primary key cannot be altered after the table is created;
• a key value cannot be changed by UPDATE or UPSERT — you delete the row and insert a new one;
• the encoded composite key is capped at 16 KiB.

The key is more than a uniqueness constraint. It is also the physical sort order of rows inside a tablet, so the column ordering you choose within the key decides which range scans get locality. A key of (device_id, ts) clusters one device's history together; (ts, device_id) clusters one instant across all devices. Both are legal, they cost the same to declare, and they perform completely differently under your real predicates.

Type coverage is narrower than Parquet's. Kudu handles the usual integers, floating point, boolean, string, binary, DECIMAL, VARCHAR and a microsecond timestamp, but CHAR and the complex types — ARRAY, MAP, STRUCT — are not supported. A schema that leans on nesting does not port without flattening. Two default ceilings are worth knowing before designing a wide table: roughly 300 columns per table, and 64 KiB per cell before encoding or compression.

Inside a tablet — MemRowSet, DiskRowSets and delta stores

The interesting question about Kudu is how columnar storage can be mutable at all, and the answer is that the columnar part is never mutated in place.

A new row arrives at the tablet leader, is appended to a write-ahead log for durability, and is inserted into the MemRowSet — an in-memory, row-oriented, key-sorted structure. When the MemRowSet passes a size threshold it is flushed into a DiskRowSet: a self-contained unit covering a contiguous range of primary keys, with each column written separately in encoded, compressed columnar form, alongside a primary key index and a Bloom filter over the keys in that range.

Once a row lives in a DiskRowSet, an update does not rewrite it. The change is recorded against that row's position in a delta store — first an in-memory DeltaMemStore, later flushed to delta files on disk. Reading that rowset therefore means reading the base columnar data and applying whichever deltas touch the rows and columns being projected. A delete is simply another kind of delta.

That leaves background work, and Kudu names each piece of it. Minor delta compaction merges several delta files for a rowset into fewer. Major delta compaction folds deltas back into the base columnar data, which is what restores full scan speed on a heavily updated rowset. Rowset compaction merges DiskRowSets whose key ranges have drifted into overlap, so a key lookup does not have to consult a growing pile of candidates.

The practical consequence is worth stating plainly: an update-heavy Kudu table whose compaction is not keeping up reads slower and slower over time, and the remedy is capacity and configuration, not a query rewrite.

Partitioning is a schema decision you mostly cannot take back

Kudu partitioning is declared in DDL and enforced by the storage engine. There is no directory convention behind it, no partition created as a side effect of a write, and nothing resembling Hive dynamic partitioning. Two schemes exist and they compose.

Hash partitioning — PARTITION BY HASH(col) PARTITIONS n — assigns each row to one of n buckets by hashing one or more primary key columns. It exists to spread writes evenly, and it is the cure for the classic time-series hot spot where every new row carries the largest timestamp and therefore lands on the same tablet.

Range partitioning — PARTITION BY RANGE(col) (PARTITION ...) — cuts the key space into explicit intervals. Ranges are not implicit: you create them, and a row whose key falls outside every defined range is rejected rather than quietly placed somewhere. Ranges can be added and dropped after creation, and dropping one deletes its data as a metadata operation over whole tablets. That is your retention mechanism, and it is enormously cheaper than DELETE WHERE ts < ..., which would generate a delete delta per row.

The canonical time-series design combines both: hash on a high-cardinality identifier for write spread, range on time for pruning and cheap expiry.

What you cannot change is the hash bucket count. It is fixed at table creation, and the only way to alter it is to create a new table and rewrite the data through it. Kudu also does not split tablets automatically — pre-splitting is manual — so the partition count you declare is the parallelism you get for the life of the table. The published guidance to plan against is on the order of 10 GiB of data per tablet, around a thousand tablets per tablet server, and about 60 tablets per table at creation time. Production clusters run well past all three, but those are the numbers that describe comfortable operation rather than a rescue project.

Tablets, Raft and the master

Each tablet is an independent Raft consensus group. The replication factor is set per table and must be odd, with three the normal choice. One replica is the leader: writes go to it, it appends them to its write-ahead log and replicates them to the followers, and the write is acknowledged once a majority has durably logged it. If the leader fails, the survivors elect a new one within seconds, and Kudu re-replicates a tablet whose replica has been missing long enough to look permanent.

Because consensus is per tablet rather than per server, any given tablet server is leader for some of its tablets and follower for others, so write leadership spreads naturally across the cluster. It also means a single node failure removes a fraction of the leaders rather than a whole shard of the table, and recovery is many small elections instead of one large reassignment.

The master holds the catalog: table schemas, the mapping from hash buckets and range intervals to tablets, and which tablet servers hold which replicas. Masters are themselves Raft-replicated — three is the standard deployment — and they are deliberately kept out of the data path. A client asks a master where a tablet lives, caches the answer, and then talks directly to tablet servers.

Impala works exactly this way. The frontend resolves the table's partitioning and the query's predicates into Kudu scan tokens — one per tablet, each carrying the column projection and the pushed predicates — and the scheduler assigns each token to an executor, preferring one co-located with a replica of that tablet. Tablet count is therefore also your scan parallelism, which is the second reason the partitioning decision above matters.

Baseline scaling guidance is three masters and around a hundred tablet servers, at roughly 8 TiB of post-replication data each. Clusters of three hundred servers and beyond are reported in production, but the smaller figures are the supported envelope.

Creating Kudu tables from Impala — STORED AS KUDU, internal and external

The DDL is ordinary Impala SQL with two Kudu-specific clauses: an inline PRIMARY KEY and a PARTITION BY specification.

CREATE TABLE sensor_readings (
  device_id  BIGINT,
  ts         TIMESTAMP,
  metric     STRING,
  value      DOUBLE,
  quality    TINYINT,
  PRIMARY KEY (device_id, ts, metric)
)
PARTITION BY HASH(device_id) PARTITIONS 16,
             RANGE(ts) (
  PARTITION '2026-07-01' <= VALUES < '2026-08-01',
  PARTITION '2026-08-01' <= VALUES < '2026-09-01'
)
STORED AS KUDU;

Note PARTITION BY, not the PARTITIONED BY that HDFS tables use. It is a different keyword for a different mechanism, and it is a reliable first-day error.

Impala needs to know where the Kudu masters are. Normally that comes from the --kudu_master_hosts startup flag on impalad (host and port 7051, comma-separated for high availability); if the flag is unset, each table carries TBLPROPERTIES('kudu.master_addresses' = '...') instead, which is workable but leaves cluster topology scattered across table metadata.

Internal versus external. A table Impala creates and manages is internal: DROP TABLE really drops the Kudu table and its data. The underlying Kudu table is given a generated name of the form impala::db_name.table_name, so two Impala databases cannot collide in Kudu's flat namespace. An external table instead maps a Kudu table created elsewhere — by the Java or Python client, or by a Spark job:

CREATE EXTERNAL TABLE readings_map
STORED AS KUDU
TBLPROPERTIES ('kudu.table_name' = 'my_kudu_table');

Dropping that removes only the mapping; the Kudu table survives. Creating external Kudu tables is a privileged operation — it requires ALL on SERVER under Ranger, precisely because it lets a user attach a SQL name to arbitrary underlying storage.

Where Kudu's Hive Metastore integration is enabled, Kudu registers its own tables in the same metastore Impala reads, so tables created outside Impala become discoverable instead of needing a hand-written mapping; only internal tables are synchronized automatically. ALTER TABLE covers what the storage engine allows: adding and dropping non-key columns, renaming, changing a column's default, encoding or compression, and adding or dropping range partitions — never the key and never the hash bucket count.

INSERT, UPDATE, DELETE and UPSERT — what actually reaches Kudu

This is the part with no equivalent on an HDFS table. Against Kudu, Impala supports all four, and each is a genuine row-level operation carried to the relevant tablet leaders by the Kudu client rather than a file rewrite:

• INSERT adds rows, honouring uniqueness and NOT NULL on the key;
• UPDATE changes non-key columns of existing rows;
• DELETE removes rows;
• UPSERT inserts where the key is absent and updates the non-key columns where it is present.

The duplicate-key behaviour is the detail to internalise, because it is not what a database background prepares you for. By default, an INSERT that collides with an existing primary key does not fail the statement. The offending rows are rejected, every other row is still inserted, and the statement completes with a warning. A pipeline that treats "no error" as "all rows landed" will drop data silently. Either check the warning count, or use UPSERT, which makes the operation idempotent and is the correct default for change-data-capture feeds and late-arriving corrections.

The same model means a DML statement is not atomic by default. There is no rollback across the rows of one statement; each write commits in its own tablet as it lands, so an interrupted statement leaves a partially applied result. Impala does offer multi-row transactions for INSERT and CREATE TABLE AS SELECT behind the ENABLE_KUDU_TRANSACTION query option, and inside a transaction the duplicate-key case aborts rather than warns — a meaningful semantic switch to be aware of. But the design that ages well is idempotent UPSERT over a primary key that encodes the natural identity of the record, so replaying a batch is harmless.

Two traps on the write path. Primary key columns cannot be changed by UPDATE or UPSERT at all; correcting a key is a DELETE followed by an INSERT. And Impala's 96-bit nanosecond TIMESTAMP is stored in Kudu's 64-bit microsecond column, so the nanosecond portion is rounded away on write — values written and read back are not bit-identical, which matters if a timestamp is part of your key.

Predicate and projection pushdown, and how to read it in EXPLAIN

Two things make a Kudu scan cheap, and both are visible in the query plan.

Projection pushdown is automatic and free. Kudu stores each column separately, so a scan projecting four columns of a sixty-column table reads four columns' worth of blocks. This is the same economics as Parquet, and it is the reason Kudu can be called an analytical store at all rather than a row store with a SQL front end.

Predicate pushdown is where you have to look. Kudu evaluates a bounded vocabulary itself — equality and range comparisons, IS NULL and IS NOT NULL, IN-lists — and anything outside that vocabulary stays in Impala as a residual filter applied to rows Kudu has already shipped over the network. EXPLAIN distinguishes the two explicitly on the SCAN KUDU node: kudu predicates: are the conditions Kudu applies, plain predicates: are the ones Impala applies afterwards.

EXPLAIN SELECT x, y FROM kudu_table
WHERE x = 1 AND y NOT IN (2,3) AND a IS NOT NULL
  AND b > 0 AND length(s) > 5;

  predicates: y NOT IN (2, 3), length(s) > 5
  kudu predicates: a IS NOT NULL, b > 0, x = 1

Read that split as a bill. The NOT IN and the function call cost a full row transfer for every row that satisfied the pushed predicates, however few survive afterwards. Rewriting a filter into a form Kudu understands, or materialising a computed value into a stored column so it can be predicated on directly, is a real optimisation rather than micro-tuning.

Above the per-row filters sits partition pruning. A predicate on a range-partitioned column lets Impala skip whole tablets before it ever contacts them; equality on every column of a hash dimension prunes hash buckets the same way. Inside a tablet, predicates on key columns exploit the primary key index and the per-rowset key bounds, so Kudu skips rowsets it can prove irrelevant. Impala additionally pushes minimum and maximum values derived from a join's hash table down into the Kudu scan as a runtime filter, so a selective dimension table narrows the fact-table scan without you writing the predicate yourself.

The metadata story — why Kudu tables skip most of the REFRESH problem

On an HDFS table, Impala caches the file listing and block locations, and anything that writes files outside Impala stays invisible until somebody runs REFRESH. That single fact is responsible for a large share of Impala's operational folklore and a good number of its 3 a.m. pages.

Kudu tables largely do not have that problem, and the reason is structural rather than clever: there is no file listing to cache. Impala enumerates nothing on disk for a Kudu table; it asks the Kudu client for rows at query time. So data inserted, updated or deleted through any client — a Spark job, the Kudu Java API, a different Impala cluster — is visible to the next query with no REFRESH and no INVALIDATE METADATA. The documentation states this directly: neither statement is needed when data is added to, removed from, or updated in a Kudu table, even when the change is made through the Kudu API.

What Impala does cache is the table's schema, so the exception is narrow and specific. After a schema change made outside Impala — adding or dropping a column through the Kudu API — you do need INVALIDATE METADATA (or REFRESH) on that table before Impala sees the new shape. Under Kudu's Hive Metastore integration, INVALIDATE METADATA is also how Impala discovers tables that were created in Kudu from somewhere else. Data changes: free. Shape changes: still a cache invalidation.

Two adjacent conveniences fall out of the same structure. Kudu tables have no small-file problem, because there are no query-visible files to accumulate — compaction is the storage engine's own business. And partition metadata lives in the Kudu master rather than the Hive Metastore, so a table with thousands of partitions does not load the catalog service the way a thousand-partition HDFS table does.

One thing does not change: COMPUTE STATS. The planner still needs row counts and distinct-value counts to choose join order and to decide broadcast versus partitioned exchange, Kudu does not supply them, and a Kudu table with no statistics misplans exactly like an HDFS table with none.

Read modes, clocks and the consistency you actually get

Within one tablet, Kudu is straightforward. Writes go through the Raft leader and commit on a majority, so a tablet has a single ordered history and no divergent replicas to reconcile.

Across tablets, you choose. Impala exposes the choice as the KUDU_READ_MODE query option with three values. DEFAULT defers to the impalad startup flag --kudu_read_mode. READ_LATEST gives read-committed isolation and nothing more — every row returned was committed at some point, but the scan is not a snapshot, so a query spanning many tablets can legitimately mix states from different instants. READ_AT_SNAPSHOT takes a snapshot of the current state and scans that, which delivers read-your-writes consistency within an Impala session, with the documented caveat that a Kudu leader change can break the guarantee.

Snapshot reads are not free. To serve a consistent timestamp, a tablet has to wait for operations that were in flight at that timestamp to resolve, so scans can block briefly under heavy write load. That is the trade being made, and for a dashboard reading a table under continuous upsert it is usually the right one — a report that silently mixes two states of the same entity is a worse outcome than a report that takes an extra second.

The clock is a hard dependency. Kudu's MVCC timestamps rely on the machine clock being synchronised. With the default system time source, a master or tablet server whose clock is reported unsynchronised, or whose error bound exceeds the --max_clock_sync_error_usec threshold (10 seconds by default), refuses to start and logs a fatal error. Since Kudu 1.6 a running server tolerates brief losses of synchronisation, but it will not survive hours of drift. NTP is not hygiene on a Kudu cluster; "the tablet servers will not start" is one of the most common first-deployment failures, and the cause is almost always time, not Kudu.

Kudu against Iceberg and the open table formats

For a while, Kudu was the only credible way to have a mutable analytical table on Hadoop. That is no longer true, and being honest about the comparison is more useful than defending the technology.

Iceberg, Delta Lake and Hive ACID all achieve row-level update and delete over immutable files on HDFS or object storage: a write produces new data files plus delete files, readers resolve the two, and compaction periodically folds them together. What that model buys is considerable. The data sits in cloud object storage, so storage and compute scale and are billed separately. The same table is readable by Impala, Hive, Spark, Trino and Flink through a shared open specification, so no single engine owns it. And snapshot isolation, schema evolution and time travel arrive as table-format features rather than engine features. The mechanics of all that belong on the Iceberg pages for Spark and Hive and are not repeated here.

What the file-based model cannot deliver is per-row latency. A commit in an open table format is a metadata operation that produces files, so the natural granularity is a batch every few minutes, and the sustainable commit rate is bounded by how much metadata churn and compaction you are willing to run. Kudu's granularity is one row, immediately, because it is a running service with memory, a write-ahead log and Raft consensus rather than a layout convention over object storage — and that row is queryable by the very next Impala scan.

That is the whole distinction, and it is the one to decide on. If mutations arrive as batches and you want object storage plus multi-engine access, the open table formats have won that ground decisively; choosing Kudu there means operating a stateful storage cluster to buy latency you do not need. If you have a continuous stream of individual upserts against a table that must also serve interactive analytical scans seconds later, Kudu is still the component that does it — on local disks, in a cluster you run yourself.

When Kudu is the wrong answer

Kudu is a specialist, and most tables are not its speciality. The failure mode is not that Kudu breaks; it is that you take on a stateful distributed storage system to solve a problem that did not require one.

Append-only analytics. If rows are written once and never corrected, Kudu is pure overhead — you are paying for a primary key index, delta stores, compaction and a replication service to support mutations you never perform. Parquet on HDFS or object storage is cheaper, faster to scan and dramatically simpler to operate.

Cloud and separated storage. Kudu stores data on the local disks of its tablet servers. There is no object-storage backend and no separation of storage from compute. On a cloud migration this is usually the argument that ends the discussion, however well Kudu fits the access pattern.

High-cardinality random access. HBase still wins where the workload is millions of point operations per second against a sparse, schema-flexible row model. Kudu's fixed schema, roughly 300-column ceiling and 64 KiB cell cap are simply not built for wide, sparse rows.

Anything needing indexes other than the key. The primary key is the only index. A selective predicate on a non-key column is pushed down, but it is still evaluated across every row the partition pruning left behind. If the access pattern is "look this up by any of five columns", Kudu will scan for four of them.

Transactional workloads. No foreign keys, no secondary indexes, no general multi-statement transactions, and no SQL of Kudu's own — the query engine is always something else. It is not an OLTP database with a fast scan bolted on the side.

Very large clusters or nested data. The comfortable envelope is on the order of a hundred tablet servers at single-digit TiB each, and a schema that depends on ARRAY, MAP or STRUCT does not port at all without flattening it first.