从批处理迁移到微批次流式处理的实战经验

简介 许多被描述为批处理系统的数据管道，实际上都已经是以近乎不间断的方式运行。它们频繁地处理增量数据——通常采用重叠窗口机制，其主要作用是缩小两次大规模、低频率全量重新计算之间的数据时效性差异。 本文描述了这样一个系统的迁移过程。该系统包含一组定时执行的批处理作业，负责生成用于搜索和广告检索管道的增量索引。 这些作业已经迁移到一个持续运行的微批处理模型，使用了基于 Spark Structured Streaming "的微批处理模式，其目的不是实时地处理每条记录，而是为了消除调度延迟，提高运维可预测性。 虽然Spark Structured Streaming是作为执行引擎，但进度跟踪并未依赖其原生的检查点机制或事件时间watermark语义，因为管道是根据分区的进度推进，而不是基于连续的事件流。相反，系统在外部维护了一个逻辑watermark，基于分区时间戳来表示最新处理过的分区。 该系统处理存储在对象存储（S3风格）中按时间分区的数据，而不依赖于Kafka等事件流。虽然曾经考虑过基于日志的摄入模型，但该管道处理的是面向批处理的快照式数据，分区层面的完整性比每条记录的顺序更为重要，因此，基于对象存储的摄入方式更为合适。 因此，进度取决于对分区数据的列举和解析，而非处理有序事件。 关键挑战不在于计算效率，而在于如何在以下环境中保证进度的稳定可靠： 数据以时间分区文件的形式到达。完成信号并不可靠。数据的新鲜度比严格重放中间状态更为重要。 该系统并未采用记录级流式处理，而是最终采用了一个基于以下内容的更简单的模型： 基于时间的执行。基于时间戳的进度跟踪。始终仅处理最新的可用分区。 本文重点探讨了本次转型中的实践经验：哪些做法失败了，哪些行之有效，以及这些设计选择适用哪于些限制条件。 系统范围和用例 这项工作针对的是负责摄取新广告、营销数据以及产品、商品及客户信号（如转化率、表现和联购行为）的生产级批处理任务。这些输入数据经过处理后会生成供在线检索服务使用的倒排索引。该系统处理的文档规模达数百万份，完整索引的大小约为数百GB，增量索引的大小通常有数十GB。 这些任务并非通用的流式处理基础设施，它们是功能固定的处理管道，对对象存储中按时间分区的数据进行处理，不涉及事件流或按记录处理的语义。它们在时效性关键路径上，一旦出现延迟，就会导致更新后的广告及相关元数据上线推迟，从而引发检索结果过时，甚至在最新版本投入生产前耗尽广告预算，进而导致错失商机。 在状态稳定的情况下： 新的增量数据大约每五到七分钟更新一次，具体时间取决于新广告的数量以及现有营销活动的最新情况。每次增量更新大致涵盖过去五小时的数据。每小时进行多次增量更新既实用也符合预期。 尽管任务数量不多，但它们的行为会直接影响数据的新鲜度。主要的瓶颈并非计算本身，而是调度延迟和协调开销，尤其是在有突发负载和故障时。 背景：全量索引和增量索引管道 我们的索引系统由两个职责截然不同的管道组成。 全量索引管道会从头重建整个Solr索引，其中包含所有广告和元数据，以及供下游相关性与质量模型使用的商品、商品详情、客户、转化、展示次数和行为信号（如“一起购买”）。这个过程资源消耗大、成本高、耗时长，因此无法频繁执行。一次完整重建大约需要两到三个小时，随后还需要进行验证和部署，总耗时可达约五个小时。部署过程采用了全索引交换方式（可以确保更新操作是原子性的），而非通过部分段更新来实现。 为了在两次完整索引运行之间将变更推送到生产环境，我们采用了增量索引管道。该管道仅处理广告、广告组和广告活动的增量更新，而且刻意保持一个比较小的规模。一个典型的增量索引大小约为完整索引的十分之一，因此可以频繁地重新生成。 理论上讲，这种方法本应加快更新的传播。但在实际应用中，增量管道是由外部调度的，每次都作为独立的作业调用来执行。随着时间的推移，逐渐显现出来一些问题。在计划运行刚结束时到达的增量数据，往往需要等待几乎一个完整的调度周期才会被处理。由于进度是在作业级跟踪的，所以即使只有部分任务未完成，一旦发生故障，也必须重新执行整个计划时间窗口内的所有任务。在更新高峰期，批处理时长会增加。在外部调度模型中，这种增加会缩短甚至消除两次运行之间的空闲间隔，甚至可能导致计划任务被跳过或延迟。这样一来，进度不仅受处理时间限制，还受固定调度边界限制，导致数据新鲜度滞后的问题加剧。 问题的核心并不在于批处理本身，而在于粗粒度的调度边界与作业级进度语义的组合。最终，人们逐渐认识到，造成数据新鲜度滞后的主要原因并非处理成本，而是批处理调度延迟和协调开销。 流式处理为何在公司内部引发了争议 当流式处理首次被提出时，反对的声音并非源于其正确性或性能，而是源于运维方面。 资深工程师们提出了一些完全合理的担忧： 长期运行的流式处理任务在运维层面更难把握。恢复行为可能难以预测。故障往往会持续存在，而非干净利落地终止。值班负担往往会增加，而非减轻。 实际上，其中一些问题确实出现了，特别是在长期运行的作业行为和内存方面；而当我们简化了系统的执行流程和重启处理机制后，其他问题则不再那么突出。 这些问题至关重要。我们的目标绝不是用一个运行机制不透明、会出现新型故障且更难调试的系统，来取代可预测的批处理作业。 任何向流式处理的转型，都必须直接解决这些运营风险。 错误的开始：记录级流处理 与许多团队一样，我们最初采用的是记录级流处理（record-level streaming）。从理论上讲，这种方法看起来是最“正确”的解决方案。但在实践中，我们很快意识到，对于这条数据管道而言，这种做法既没有必要，又存在风险。 索引逻辑做了批次完整的假设，而且是在产品或商品分组层面进行操作，而非在单条广告记录的层面。这种分组主要源于中间处理管道的架构选择，数据会在更高层级的实体中进行聚合，然后在最终索引中展开回单条记录。通常，一条广告的变更需要重新计算增量索引中该产品或商品的分组表示。若改用记录级流式处理，将会引入部分更新状态——即部分广告已更新，但分组索引表示尚未完全一致。实际上，这种做法要么迫使整个受影响的增量分区重新生成，要么冒着搜索结果暂时不一致的风险。 针对这一问题进行重构将需要做大量的改动，而且收益不明。更重要的是，业务并不需要每条记录都有很好的即时性。它真正需要的是不用等待批处理调度完成。这是第一个重要的认识：记录级流式处理正在解决一个我们并不存在的问题，同时引入了我们不希望出现的问题。 收敛于微批次流式处理 我们是有意转向微批次流式处理的。我们的目标不是连续地处理记录，而是实现持续可用性。微批次处理使我们能够消除调度间隙，并在关键环节保留批处理语义。 在运维层面，作业被配置为约三十秒的固定触发间隔。该配置是通过Spark Structured Streaming的微批次处理模式实现的，并且采用了固定的处理时间触发机制。我们特意将触发间隔设置得远小于五到七分钟的分区到达频率，为的是确保新数据能被迅速捕获，而且不存在外部调度间隙。 每次触发操作都按照一个有界且确定的顺序进行： 确定当前的挂钟时间。根据时间和分区规则，计算应被视为符合条件的最新分区。将该分区与当前的watermark（最后一个已确认的分区）进行比较。如果有多