Why Message Queues Can't Do Real-Time Aggregation: Alibaba Mama's Flink + Paimon Lakehouse

1. Problem Background and Technical Evolution

Advertising is a "data-as-product" industry: advertisers monitor impressions, click-through rates (CTR), and ROI in real time to decide whether to increase budgets or pause campaigns. This creates three hard requirements distinct from risk control:

External-facing, money-related : Advertisers complain immediately if they see duplicate or delayed data; updates delayed by ten-plus minutes can be perceived and escalated.

Promotional tidal spikes : During 618, Double 11, and Double 12, both traffic and query loads surge simultaneously.

Massive scale : Daily incremental data reaches trillions of records, TPS in the tens of millions, with requirements for fast failure "stop-the-bleeding" and support for canary rollback.

Early needs were often met with a "message queue + Flink" quick-launch approach, but as business lines grew, issues like duplicate consumption, inconsistent metrics, and inability to build an aggregation layer erupted. Alibaba Mama's evolution fully illustrates why message queues cannot sustain a real-time data warehouse and why Paimon's upsert capability is the key missing piece.

2. What You Will Gain from This Article

Understand the three structural defects of message-queue-based (TT/Kafka-like) real-time data warehouses in deduplication, DWS aggregation layer, and schema .

See how Paimon primary-key table upsert + Changelog fills the gaps for DWD deduplication and DWS aggregation coverage.

Grasp the design and query arbitration logic of the offline rewrite correction pipeline (anti-fraud/correction written back to the lake).

Learn a set of production optimization practices from Alibaba Mama: async compaction, checkpoint interval tuning, bucket/partition pre-planning, file lifecycle management, and Consumer configuration.

Distinguish advertising reporting scenarios from risk control scenarios in latency targets (minute-level vs. sub-second) and storage model selection .

Be able to answer interview questions like "Why does a real-time data warehouse need upsert? Why can't a pure message queue handle the aggregation layer?"

3. Core Concepts and Architecture Positioning

The data sources remain unchanged (behavior logs, ad logs, conversion data collected via log ingestion/DRC into TT — a Kafka-like message queue). The key change is in the processing and storage layer :

DWD layer : shifted from "write to TT, downstream parses independently" to using Paimon primary-key tables .

New DWS layer (absent in the old architecture because it couldn't be built) also uses Paimon.

DWD/DWS Changelog for downstream real-time consumption; tables themselves have schemas and can be queried directly by algorithms/BI.

High-frequency, large-scale queries ultimately land in ClickHouse ; offline anti-fraud processed data is written back to Paimon for correction.

In one sentence: "This moves DWD/DWS from message queues into lake tables that support upsert, while ODS remains a message queue." Unlike the DiDi case (Article 18) which built an online feature store, this runs the full real-time data warehouse layering (ODS-DWD-DWS-ADS) on the lake.

4. Deep Dive into Underlying Principles

4.1 Why the Old Architecture Got Stuck: Three Structural Defects

Defect 1: DWD Downstream Duplicate Data

ODS/DWD jobs re-send data due to business iterations or failover switches. Downstream consumers of DWD may read duplicate records. Whether to deduplicate at DWD depends on the business, but advertising almost always requires deduplication — otherwise advertisers see abnormal year-over-year spikes. In a message queue, deduplication must be done by each downstream consumer at high cost.

Defect 2: No DWS Layer Because Message Queues Don't Support Upsert

This is the core issue. A real-time data warehouse typically wants a DWS aggregation layer between DWD and applications (aggregation logic is relatively fixed and reusable). But message queues can only append, not overwrite : suppose a 5-minute window aggregation; after the 10:05 result is written to the queue, at 10:10 you need to overwrite the previous window's result — both records remain in the queue, forcing downstream to deduplicate again. The sharing party measured that this approach at least doubles resource consumption, and jobs become unstable at scale. Consequently, the old architecture had to write aggregation results directly into online stores like OceanBase, Hologres, or ClickHouse — inflexible and with insufficient real-time anti-fraud precision.

Defect 3: No Schema

Data between ODS and DWD is semi-structured; after parsing there is no fixed schema. Downstream algorithms and BI must each re-parse and re-compute, duplicating effort. On top of that, a Lambda offline pipeline (ODS synced to MaxCompute for ETL, aggregated into ADS, then written back to online stores) meant two separate codebases, doubling development and maintenance costs.

4.2 Why Paimon Fills the Gap: Primary Key Upsert + Changelog + Schema

Alibaba Mama began researching Paimon in June 2023. Its mechanisms address the defects directly:

Primary-key table upsert : At DWD, upsert writes achieve deduplication; at DWS, upsert overwrites the latest aggregation result for the same time window — turning "two records in the message queue" into "only the latest value per primary key is retained," which is the LSM primary-key model capability described in Article 10. Downstream consumes Changelog (change stream) to get increments without re-deduplication.

Unified schema : DWD/DWS tables have fixed schemas, allowing algorithms, BI, and operational data to query these two layers directly in both real-time and offline modes, eliminating repeated parsing.

Single codebase for streaming and batch : Unified schema enables stream/batch reuse, removing the dual-maintenance Lambda architecture.

4.3 Offline Rewrite Correction: Advertising-Specific Pipeline

Advertising has real-time anti-fraud, but its precision is insufficient; final correction relies on offline processing. Alibaba Mama retained an offline pipeline: parsed ODS data can be written back to Paimon in reverse . At query time, arbitration occurs — if a correction exists, query the corrected data; otherwise fall back to real-time data . This pipeline also serves as a backup: when Flink fails, offline data provides a safety net. Hence the overall design is "two real-time pipelines + one offline pipeline" with daily switchability for high availability.

5. Core Pipelines and Key Flows

Main pipeline : TT (behavior/ad/conversion logs + dimension table increments) → Flink ETL → Paimon DWD (primary-key upsert deduplication) → Flink aggregation and upsert write → Paimon DWS (overwrite latest aggregation for same window) → high-frequency query sync → ClickHouse ; algorithms/BI query DWD/DWS directly.

Side pipeline : Offline anti-fraud processing → rewrite Paimon for correction , query arbitration follows "correction if present, else real-time." Storage layer uses primary/backup dual systems with independent sync jobs for cross-region disaster recovery and canary switching.

6. Real Business Scenarios and Engineering Practices

The sharing party provided a set of production optimization lessons (all based on their experience; adjust for your own traffic):

Write throughput : Enable async compaction per scenario — it significantly improves single-table write throughput and stability. One job saw node-switch average time drop from 50+ seconds to ~20 seconds after enabling.

Checkpoint interval must be tuned to traffic : At 100K-level traffic no tuning needed; at million/ten-million level, frequent checkpoints become a bottleneck and cause backpressure — interval should be lengthened appropriately. Writer node parallelism set by bucket count; memory increased for high-traffic scenarios.

Hard constraints before table creation : Partition and bucket count adjustments after creation (e.g., changing bucket count) require a Rescale process, possibly stopping the job — high cost. Therefore, plan partition granularity and bucket count based on peak traffic estimates upfront; don't create tables arbitrarily and change later.

File governance : Checkpoints generate many small files; at high traffic, partition/snapshot/changelog lifecycle management is mandatory. Use independent compaction jobs to merge small files; write cleanup scripts for orphan files from historical failures. Enable Consumer (records consumption offsets, retains for a period) for failure recovery, but trade-off: files retained longer, replay scans more snapshots, slower — a balance between recoverability and replay speed.

Configuration example (self-reported, specific load) : One job writes 700GB+/hour, configured with one partition per hour, 512 buckets per partition, 7-day lifecycle, supporting real-time streaming read/write; sharing party claims TPS reaches millions. These numbers heavily depend on data distribution and hardware; use only as order-of-magnitude reference.

7. Performance, Cost, and Technology Selection

Mechanism-level gains: primary-key upsert eliminates downstream deduplication; unified schema eliminates duplicate parsing; layering eliminates duplicate consumption — these are deterministic architectural benefits.

Self-reported gains (first-party data, not directly portable): Business-wise, layering reduced downstream iteration to minimal Group By/Join code, interaction cycle shortened ~30%+; DWD/DWS data directly usable for insights, one batch produced in ~15 minutes, overall latency reduced ~2.5 hours. Technically, removing duplicate consumption pipelines cut overall resource spend 40%+, development effort ~50%; dual-pipeline design supports ~99.9% availability (downtime <9 hours/year); also reduced duplicate TT subscription fees. These figures are bound to Alibaba Mama's traffic scale and cluster config; migration requires re-testing in your own environment.

Selection conclusion: Real-time data warehouses needing minute-level external reporting and "same-window aggregation overwrite" must choose a lake table supporting upsert/primary-key (Paimon-style LSM); pure append-only message queues cannot build a DWS layer. High-frequency, large-scale concurrent queries should be pushed down to OLAP engines like ClickHouse/StarRocks; sub-second online point lookups should still use the primary-key OLAP feature store model from Article 18.

8. Production Issues and Troubleshooting Methods

Issue 1: Frequent job failover, slow node switching under high traffic

Follow "phenomenon → metrics → principle → locate → fix": first check if synchronous compaction drags writes, enable async compaction; then check if checkpoint too frequent causing backpressure, lengthen interval per traffic; finally check writer memory/parallelism adequacy. Prevention: pre-estimate bucket and partition at table creation based on peak traffic.

Issue 2: Small file explosion, query slowdown

Root cause: excessive lifecycle + high-frequency checkpoints. Fix: shorten unnecessary lifecycles, independent compaction to merge small files, clean orphan files. Prevention: include file count in monitoring alerts.

Issue 3: Slow replay or state unrecoverable after failure

Root cause: default snapshot retention too short, or Consumer retention causes replay to scan many files. Fix: set Consumer retention per business need, balance recoverability vs. replay speed; offline rewrite pipeline as fallback. Prevention: dual-pipeline/offline backup, pre-drill switching before major promotions.

Issue 4: Advertisers see duplicate data

Root cause: DWD not deduplicated at primary-key layer, or failover re-sends not correctly handled by downstream Changelog. Fix: verify upsert primary-key model and Changelog semantics, dual-run reconciliation.

9. Advanced Interview and Architecture Thinking

Q1: Why can't a pure message queue do a real-time DWS aggregation layer?

Key points: Message queues only append, cannot overwrite; second aggregation for same window cannot overwrite old value, forcing downstream to re-deduplicate, doubling resources and instability; upsert/primary-key capability is required.

Q2: What three things did Paimon solve in Alibaba Mama's pipeline?

Key points: DWD upsert deduplication, DWS upsert overwrite aggregation, unified schema enabling stream/batch and downstream reuse.

Q3: Why keep the offline rewrite pipeline?

Key points: Real-time anti-fraud precision insufficient; offline correction written back, query arbitration "correction if present"; also serves as Flink failure fallback and major-promotion backup.

Q4: Why can't bucket count be changed arbitrarily after table creation?

Key points: Changing bucket count post-creation requires Rescale, possibly stopping the job; must plan based on peak traffic upfront.

Q5 (comparison): How do storage selections differ between advertising reporting and risk control features?

Key points: Advertising needs minute-level external reporting and aggregation overwrite → upsert-supported lake table + OLAP pushdown; risk control needs sub-second high-concurrency point lookup → primary-key OLAP feature store. Different latency targets drive different model choices.

10. Summary, Knowledge Extraction, and Next Episode

This article dissected Alibaba Mama's full evolution from "TT message queue + Flink" to "Flink + Paimon streaming lakehouse": core is using primary-key upsert to plug DWD deduplication and DWS aggregation coverage, unified schema to eliminate duplicate parsing, and offline rewrite to balance correction and high availability. We understood the three structural defects of message-queue warehouses, Paimon's targeted mechanisms, a set of landable production optimizations (async compaction, checkpoint interval, bucket/partition pre-planning, file lifecycle, Consumer), and clarified the model selection differences between advertising reporting and risk control features.

One-sentence conclusion: "Whether a real-time data warehouse can have a DWS layer depends on whether storage supports upsert; pure append-only message queues can only 'stream,' primary-key lake tables can 'warehouse.'"

Next episode (Article 20, Industry Deep Dive 04) enters local life / instant retail : dissecting Ele.me's real-time lakehouse exploration with Flink + Paimon + StarRocks, and Taobao Instant Retail (Flash Purchase) using Apache Fluss for second-level real-time decisions (Fluss official blog), focusing on how "second-level ultra-real-time" and "minute-level lakehouse" layer and cooperate in the same pipeline.