How to Scale an Integration Pipeline Without Breaking Correctness

如何在不破坏正确性的前提下扩展集成流水线

Data Engineering: A production account of the throughput work behind a 16x jump — and the two guarantees it was never allowed to trade away 数据工程：关于吞吐量提升 16 倍背后的生产实践，以及绝不能妥协的两大准则

Yuelin Ou | Aug 19, 2026 | 17 min read Yuelin Ou | 2026年8月19日 | 17分钟阅读

I went back and forth for a while on whether to write this up at all. The work is enterprise data integration: wiring the data from a lot of separate business systems together through a pipeline. Orders, inventory, finance, logistics, customer records, plus a pile of legacy FTP batch channels nobody wants to touch. More than twenty systems on the two ends of it. A few million events a day, several times that at month-end close and during big sales pushes. 我曾反复纠结是否要写下这些内容。这项工作属于企业数据集成：通过流水线将许多独立的业务系统数据串联起来。涉及订单、库存、财务、物流、客户记录，以及一堆没人愿意触碰的遗留 FTP 批处理通道。流水线两端连接着二十多个系统。每天处理数百万个事件，在月末结算和大型促销期间，流量更是会翻几倍。

It sounds simple. System A calls system B’s API, what’s the big deal. Anyone who has actually done this knows the annoying part isn’t getting A to talk to B. It’s keeping it correct after it’s talking. Of the twenty-odd systems, some are new and speak REST, some were outsourced ten years ago and only speak SOAP, and at least one only knows how to drop a file over FTP. The stacks are all over the place and the reliability is all over the place, and when something breaks it lands on you, because you’re the layer in the middle. 这听起来很简单。系统 A 调用系统 B 的 API，有什么大不了的？但任何真正做过这件事的人都知道，麻烦的不是让 A 和 B 通信，而是通信建立后如何保持数据正确。在这二十多个系统中，有些是支持 REST 的新系统，有些是十年前外包的、只支持 SOAP 的老系统，甚至至少有一个系统只会通过 FTP 丢文件。技术栈五花八门，可靠性参差不齐，一旦出问题，责任全在你，因为你是中间的那一层。

This article is about the third of three problems that pipeline forced me to solve, and the one people usually reach for first and get wrong: throughput. The pipeline has to keep day-to-day latency under about half a second and absorb roughly ten times normal volume at peak, which in practice means tens of thousands of events a second on an ordinary peak and more than that during a sale. The trap is that almost everything you do to go faster is also a way to silently break the data, and once the data is wrong you find out about it weeks later, from finance, during a reconciliation, which is the worst possible time. So I can’t talk about speed without first being clear about the floor I wasn’t allowed to drop below. 本文要讨论的是该流水线迫使我解决的三个问题中的第三个，也是人们通常最先尝试却最容易出错的一个：吞吐量。流水线必须将日常延迟保持在半秒以内，并在高峰期吸收大约十倍于平时的流量。在实践中，这意味着在普通高峰期每秒要处理数万个事件，促销期间则更多。陷阱在于，几乎所有为了提速所做的优化，都可能悄无声息地破坏数据。一旦数据出错，你往往要等到几周后财务对账时才会发现，而那是情况最糟糕的时候。因此，在谈论速度之前，我必须先明确我绝不能逾越的底线。

A note on where the numbers come from

关于数据来源的说明

Before any of the figures below, it’s worth being honest about what kind of numbers they are. Everything I quote is a consumer-side runtime metric taken from the live pipeline during normal operation, not a controlled benchmark on a clean cluster. Throughput is events processed per second measured at the consumer, read off during ordinary business-hour traffic rather than at peak; when I say a rate was “stable” I mean it held within normal variance across full business cycles, not that I pinned it in a single run. The batch-size comparison later (50, 100, 200, 500) was run against real production load, not synthetic data, which is why the answer is specific to this workload and not a universal constant. 在列出以下数据之前，有必要诚实地说明这些数字的性质。我引用的所有数据都是在正常运行期间从实时流水线的消费者端获取的运行时指标，而非在干净集群上进行的受控基准测试。吞吐量是指在消费者端测得的每秒处理事件数，读取自正常的业务工作时间而非高峰期；当我提到某个速率“稳定”时，是指它在完整的业务周期内保持在正常波动范围内，而不是指在单次运行中锁定的数值。稍后提到的批处理大小比较（50、100、200、500）是针对真实的生产负载而非合成数据运行的，这就是为什么结论仅适用于当前工作负载，而非通用常数。

Where a figure is softer than it looks, I say so. These numbers were collected across multiple month-end close and peak-sales cycles of normal operation, not in a single benchmark run. I’m reporting an experience, not a study, and the value of it is in the failure modes and the trade-offs, not in a benchmark you could rerun. 如果某个数字比看起来更“软”（不精确），我会说明。这些数字是在多个月末结算和促销高峰的正常运行周期内收集的，而非单次基准测试。我是在分享经验，而非撰写研究报告，其价值在于故障模式和权衡取舍，而不是你可以重复运行的基准测试。

The floor: what scaling is not allowed to break

底线：扩展绝不能破坏什么

Two guarantees sat underneath every throughput change, and every one of the optimizations later in this article is built so it can’t violate them. The first is that a later version of an entity’s state can never be overwritten by an earlier one. In a distributed pipeline the same logical update arrives more than once and out of order, all the time. Network retransmits, queue redelivery, a consumer restart mid-flight, an upstream timeout-and-resend. You can’t stop any of that from happening, so the only move is to make the write path indifferent to it. 在每一次吞吐量变更的背后，都有两大准则作为支撑，本文后续的所有优化都必须确保不违反它们。第一，实体的后续状态版本绝不能被更早的版本覆盖。在分布式流水线中，同一个逻辑更新总是会多次到达且顺序混乱。网络重传、队列重新投递、消费者中途重启、上游超时重发，这些情况你无法阻止，因此唯一的办法就是让写入路径对这些情况“免疫”。

Every entity carries a version number that the source system owns (not one the pipeline invents, because the pipeline has no idea when the source actually changed something), and the write rejects anything stale: 每个实体都携带一个由源系统拥有的版本号（而不是流水线生成的，因为流水线无法得知源系统何时真正发生了变更），写入操作会拒绝任何过时的数据：

public void upsertWithVersionCheck(EntitySync sync) {
    int updated = jdbcTemplate.update(
        "UPDATE entity_store SET data = ?, version = ?, updated_at = NOW() " +
        "WHERE entity_id = ? AND entity_type = ? AND version < ?",
        sync.getData(), sync.getVersion(), sync.getEntityId(), sync.getEntityType(), sync.getVersion()
    );
    if (updated == 0) {
        // either a brand-new row to INSERT, or an older version we should drop
        try {
            jdbcTemplate.update(
                "INSERT INTO entity_store (entity_id, entity_type, data, version) " +
                "VALUES (?, ?, ?, ?)",
                sync.getEntityId(), sync.getEntityType(), sync.getData(), sync.getVersion());
        } catch (DuplicateKeyException e) {
            // a newer version already landed; dropping this one is correct
        }
    }
}

It’s basically a stripped-down last-write-wins where “last” means highest version, not most recent arrival. That one rule is what lets me be aggressive about parallelism later without lying awake about ordering. 这本质上是一种精简版的“最后写入胜出”（last-write-wins），其中“最后”指的是版本号最高，而不是到达时间最近。正是这一条规则，让我后续能够激进地使用并行处理，而无需为顺序问题彻夜难眠。

The second guarantee is that “did we already process this?” can never be wrong. Every accepted record writes its dedup-log entry and its business data in the same database transaction, so they commit together or not at all. The dedup log is the single source of truth for what was accepted, and it isn’t allowed to drift from the data it claims to describe. Early on we did the dedup check up in the business code, query first then write, and at high concurrency the gap between the two let duplicates slip through. The fix was to push it down to a primary-key constraint and let the database tell us. (That log table grows forever if you let it; a nightly job trims entries older than thirty days, which is generously past the window where redeliveries actually happen.) 第二个准则就是“我们是否已经处理过此条记录？”的判断绝不能出错。每一条被接收的记录都会将其去重日志条目和业务数据写入同一个数据库事务中，确保它们要么同时提交，要么都不提交。去重日志是判断记录是否已被接收的唯一事实来源，它绝不允许与所描述的数据产生偏差。起初，我们在业务代码中进行去重检查（先查询后写入），但在高并发下，两者之间的时间差导致重复数据漏网。解决方法是将其下沉到数据库的主键约束中，让数据库来告诉我们结果。（如果不加管理，该日志表会无限增长；我们通过一个夜间任务清理超过 30 天的条目，这个时间窗口已经绰绰有余，涵盖了所有实际可能发生的重投递情况。）

I’m spending these few paragraphs on correctness because everything below trades against it, and the trades are only safe because this floor holds. 我花这几段篇幅讨论正确性，是因为下文的所有内容都是在与它进行权衡，而只有在守住这条底线的前提下，这些权衡才是安全的。

Partitioning, and the entity that’s a hundred times louder than the rest

分区，以及那个比其他实体活跃百倍的“大户”

More partitions means more parallelism, but it also means more chances for events to be processed out of order across partitions. The rule I settled on is that every event for th… 更多的分区意味着更高的并行度，但也意味着跨分区处理时事件乱序的概率增加。我最终确定的规则是，每一个针对……的事件