Apache Hudi and Fluss: how JD.com built a Lakestream

JD.com runs a large Apache Hudi lakehouse alongside a large Apache Kafka deployment. This post describes how JD links streaming data and lakehouse data with a Lakestream: fresh data lives in Fluss tables, and a tiering service continuously writes it into Hudi tables. Readers can then query both together. The Fluss community calls this storage foundation a Lakestream: it unifies fresh data in streams with historical data in open lakehouse tables.

TL;DR

• The problem: separate real-time and batch pipelines for the same data meant duplicated storage and compute and metric definitions that drifted apart. Consumers of one impression log topic had to read 195 GB/min to get the 6 GB/min of selective columns they actually needed.
• The approach: a Lakestream on Fluss and Hudi, so one copy of data serves both real-time and batch workloads.
• The results in recommendation ranking: batch compute cost down more than 80%, storage cost down more than 50%, end-to-end training-data freshness improved by more than 30%, three training-data sources merged into one, and better click-through and conversion rates in A/B tests.
• The Fluss integration: Hudi support for Fluss tiering (being contributed upstream as Fluss Improvement Proposal 24, FIP-24), server-side consumer offset management, and high availability for the Fluss Coordinator.

Challenges at JD's Scale

Our lakehouse and streaming platforms have grown over many years. The lakehouse runs on Apache Hudi, with more than 5,000 tables and more than 700 PB of data migrated into it, and a single job peaks at more than 600 million records per minute. The streaming platform runs on Apache Kafka, with more than 8,000 nodes in production, peak writes of 500 million records per second, and peak egress of 2 TB/s.

At this scale, small inefficiencies get expensive. An extra copy of the data, a few fields read but never used, or one more pipeline to maintain each turns directly into cost, stability risk, and engineering time.

Row-Based Consumption Wastes Bandwidth and Compute

The first problem is that message queues store data by row. Our impression log topic is a good example: reading all of it means consuming about 195 GB/min, but downstream jobs actually need only 6 GB/min. Most jobs use a handful of fields, yet a row-oriented queue makes every consumer pull the full message and then parse, filter, and drop fields in its own compute job.

That wastes resources in three places:

• Network: every consumer pulls the full message to use a few fields.
• Compute: deserializing and parsing unused fields burns CPU and memory.
• Pipeline load: the bigger the topic and the more consumers it has, the more this repeated reading and parsing adds up.

Separate Lakehouse and Stream Pipelines Duplicate Work

The second problem is architectural: the lakehouse and the stream are separate systems. Take training data for search and recommendation models as an example. Teams have traditionally run two pipelines side by side. The real-time pipeline quickly turns user activity into fresh training samples and features. The batch pipeline handles backfills, corrections, model training, and analysis. Running both meets different freshness needs, but it creates three problems:

• Duplicate storage. Much of the same data is stored more than once, in the message queue and in lakehouse tables.
• Duplicate compute. Both pipelines clean the data, join it, assemble features, and generate samples. Only the execution model differs.
• Inconsistent metric definitions. The two pipelines evolve separately, so field definitions, filters, error handling, and backfill logic drift apart.

Why We Adopted Fluss

The goal is to gradually bring real-time streams and the lakehouse onto the same data model, the same service interfaces, and the same way of consuming data. One copy of data serves real-time jobs through Fluss streaming and batch reads, and is also written to Hudi tables for long-term analysis, model training, and historical replay with engines such as Apache Spark. Jobs that need both use union reads, which query the Hudi table and the newest Fluss data as one table.

Case Study: Recommendation Ranking

We chose recommendation ranking as a key production use case. The redesign gradually consolidates several parallel pipelines into one copy of data, consumed in different ways. Real-time features and labels stream into Fluss first, which supports low-latency writes and minute-level online learning. The tiering service then writes the same data into Hudi tables, where Hudi's efficient backfills support batch analysis, base model training, and catch-up training.

Results

• Cost: batch compute cost down more than 80%, storage cost down more than 50%.
• Freshness: end-to-end training-data pipeline freshness improved by more than 30%.
• Engineering effort: three training-data sources (batch, catch-up, and real-time) merged into one, and two APIs (batch and real-time) merged into one.
• Business metrics: core recommendation metrics, including user click-through rate (UCTR) and order conversion rate (UCVR), improved in A/B tests.

What We Built on Fluss

Tiering Fluss Data into Hudi

We built a Hudi plugin for the Fluss tiering service, adapting the three stages of the tiering job: the source enumerator, the source reader, and the committer.

• Runtime: TieringSourceEnumerator splits each round of tiering work and assigns the splits. TieringSourceReader reads data from Fluss and writes it to the lake table. TieringCommitter commits the lake table and reports the snapshot back to Fluss.
• Lake abstraction: a common lake storage interface hides the differences between lake formats.
• Hudi plugin: Hudi plugs in as a LakeStorage implementation.

Server-Side Consumer Offsets

We added consumer offset management to the Fluss server. The offsets live in a primary key system table, consumer_offsets, keyed by group, table_id, partition_id, and bucket_num. An offset record holds the offset, leader epoch, metadata, and commit timestamp.

Coordinator High Availability

It runs on JD's internal Kubernetes platform, with a Controller daemon deployed as a StatefulSet. In each pod, the Controller runs as PID 1 and registers with ZooKeeper on startup. The pod that registers successfully becomes active and starts the Fluss Coordinator process. The other pods are cold standbys that watch the ephemeral znode and take over if it disappears.

What's Next

We are contributing Hudi tiering back to the Fluss community as FIP-24: Tiering Service to Hudi. Our next steps include adding per-user quotas, tuning offset management, and achieving feature parity with JD's Hudi stack regarding schema evolution and dynamic bucketing.