IFCO 在 Databricks 扩展大型 dbt 项目：性能、可见性与调试

摘要

IFCO运营着全球最大的可重复使用包装循环池之一，拥有数亿个周转箱和托盘。IFCO在全球拥有2,000多名员工，其中在德国雇用超过350人，大多数人工作在位于慕尼黑附近普拉赫的全球总部。该业务是一种循环池化服务：可重复使用的塑料容器（RPC）将新鲜农产品从种植者和包装商运送到配送中心和零售商，然后返回IFCO服务中心进行清洗、分拣并再次发出，覆盖50多个国家。

每个周转箱和托盘在其整个生命周期中都被追踪，为业务所依赖的关键绩效指标（KPI）提供数据：循环时间、损耗、破损、清洗成本和池规模。将数十亿条原始追踪事件转化为可信的KPI之所以困难，有三个原因：数据量巨大，其中一些数据以难以预测的方式迟到，而当迟到数据到达时，它迫使管道修正已经报告过的历史。

本文展示了IFCO的数据平台团队如何与Databricks Forward Deployed Engineering合作，使该管道更快、更便宜。转换逻辑保留在dbt中。它运行在Databricks上，其中每个增量设置都映射到具体的Delta Lake写入行为：哪些列对数据进行聚类、一次写入需要触及目标表的多少部分，以及行是被合并还是被替换。在正确的数据粒度上正确设置这些参数，将核心语义层作业的每日运行时间减少了60%以上，并使IFCO得以退役一项昂贵的夜间全量刷新。

IFCO与数据问题的形态

一个周转箱在一年中被拣选、装满、运输、返回、清洗并重复使用多次，因此IFCO需要知道每个资产在哪里以及它经历了什么。IFCO引入了语义层，将许多不同的追踪信号汇聚为资产活动的统一治理视图：周转箱经过清洗线时的条形码扫描、码头门口的RFID读取，以及报告GPS位置、附近蓝牙信标和温度的电池供电追踪器。（在本文中，“语义层”指的是这些受治理的dbt模型，它们将原始追踪事件转化为业务KPI。）有三个特性使这件事变得困难。

• 规模。每天流入数十亿条追踪事件。合并会将每个资产的重复ping压缩为少得多的活动行，但KPI所读取的表仍然大到从头重建它们代价高昂。
• 不可预测的迟到。大多数观测落在预期窗口内，但有些数据源延迟数周，少数延迟数月，且没有固定时间表。一个假设今天拥有的数据就是昨天所发生事情全貌的管道，会悄无声息地报告错误的历史。
• 历史对账。资产活动是一个序列，因此迟到的观测不仅仅填补一个空白。将一条扫描记录插入资产时间线的中间，会改变管道已经就其后所有内容得出的结论：资产接下来去了哪里、它的循环何时开始、它落入哪个KPI桶。因此，迟到数据迫使管道重新计算它已经发布的状态，而不是简单地追加一个新行。目标是快速输出一个良好的估计，并随着迟到数据的到达将其收敛到真相，而无需每晚重新处理所有内容。

大规模增量处理

dbt增量模型在底层是一组Delta读写行为，而大部分收益来自一个原则：让每次运行触及尽可能少的行，并尽早削减它们。第一个也是最大的杠杆是读取本身，只扫描一次运行实际需要的文件和变更资产，因为你避免读取的每一行都是永远不会到达下游昂贵的逐资产排序、洗牌和写入的行。下面每一项技术都是普通的dbt配置，会转化为特定的Delta行为。

在你用于过滤和连接的列上进行聚类。液态聚类，以每个模型被查询时所依据的粒度（对于资产活动，是资产和事件日期）为键，让引擎跳过文件而不是扫描它们。正是它使接下来的两项技术得以奏效。

有意识地选择增量策略。该策略决定每次运行如何写入，而选择取决于两个问题：每一行是否有稳定的键，以及你是在原地更新行还是一次替换一组行？对于有键、去重密集的upsert，merge是默认选择。以真实粒度为键（对于资产活动，是asset_id和event_date_time），它做了两件批量删除再插入无法做到的事：

等值连接谓词启用动态文件裁剪：传入批次中的键值会跳过不可能包含匹配项的目标文件，因此写入只触及它所更改的那一片。（DBT_INTERNAL_DEST和DBT_INTERNAL_SOURCE是dbt在其生成的语句中为目标表和传入批次所使用的别名。）在merge所匹配的相同键上进行聚类，可保持这种裁剪紧密。一个行哈希防护，即一个matched_condition，比较每一行的代理哈希，然后跳过重写实际上未更改的行，这节省了写入并保持下游变更流干净。

delete+insert是替代方案：它按键删除一整组行并重新插入。当一次运行将一组行作为一个单元重新推导，且这些行没有可匹配的稳定身份时，这更简单，代价是即使没有任何更改也要重写该组。在非常大的数据量下，两者值得进行基准测试，而不是想当然。

将写入限制在近期窗口内。同样的谓词机制还有第二个用途。不是用等值连接进行文件裁剪，而是用时间界限将写入限制在近期数据上，因此在数据量更大的上游模型中，MERGE匹配的是目标表的近期切片，而不是整个表：

因为谓词以行被摄取的时间为依据，而不是事件发生的时间，所以只要一个事件最近才落地，即使它已经发生数月，仍会被捕获。窗口只需足够宽，以覆盖数据落地与本次作业处理它之间的间隔。把窗口设得太窄，迟到的数据就会被静默跳过：它不会报错，只是永远得不到处理。

只重新计算发生变化的部分。模型把自身工作范围限定在受新数据或迟到数据影响的资产上，这些资产通过摄取水位线识别；并且读取比写入更宽的窗口，这样迟到的数据事件无需全量刷新就能被捕获。

保持 Delta 整洁。重度增量表会开启优化写入和自动压缩，或者把表维护交给 Predictive Optimization，这样频繁的合并就不会留下小文件读取税。

关键在于把这些措施应用到正确的粒度上，然后从实际的查询计划中确认引擎确实在进行裁剪，而不是在静默地全表扫描。

一个实例：将观测数据整合为资产活动

语义层中最繁忙的模型，是将来自每种跟踪技术的观测数据整合为每个资产的单一、位置感知流的那个模型。它使用按资产分区、按事件时间排序的窗口函数，推算出资产实际何时移动。当一条观测数据没有明确位置时，它会退回到 Databricks SQL 内置的 H3 函数，这些函数将每个纬度/经度映射到一个六边形网格单元，从而使“同一地点”变成对单元 ID 及其网格距离的低成本比较，而不是反复进行地理距离计算。它曾是迄今为止运行时消耗最大的单个模型。

第一步不是去优化，而是看清实际在运行什么，这一区别很重要。dbt compile会渲染出模型的SELECT，并解析其中的 ref，但对于增量模型来说，这并不是 Databricks 实际执行的语句。在那条编译后的SELECT背后，dbt 会生成并运行一个更大的操作：临时视图、对目标表的扫描，以及最终写回表。要找出时间和内存花在哪里，唯一的方法是从查询历史中逐个阶段地阅读实际执行的查询计划，而不是编译后的 SQL。

这样读下来，那个计划问题严重。该模型扫描了数十亿行，向磁盘溢写了数百 GB，并将约 85% 的时间花在单个按资产的窗口排序和洗牌上。实际上，它每次运行都在重建整张表。造成这一点的有三个原因：

• “已变更”资产集合从未缩小。上游时间戳每次运行都会被重新生成，而不是从源端一路传递下来，因此几乎每个资产看起来都是新的。当一切都显得脏时，增量运行就会悄悄变成全量刷新。
• 按资产重新计算没有边界。窗口函数会一直回溯每个资产的完整历史，因此即使真正发生变化的集合很小，也会把数年的历史拖入排序。
• 窗口阶段计算了最终查询从未使用的列，其中包括第二个前瞻性窗口，其全部输出都被丢弃。

每个修复都直接对应其成因：让真实的摄取时间戳穿过上游模型传递，使变更集反映真正的新数据；将重新计算绑定到近期窗口；删除未使用的列和前瞻窗口；按模型被查询的粒度进行聚簇。最后，将整个图作为并行的按模型任务运行在无服务器计算上（见下一节）。这些措施合在一起，将核心作业的运行时减少了 60% 以上，接近三分之二，并取消了此前为保持 KPI 正确而必须进行的夜间全量刷新。

从一次调试会话到一项可重复的技能

上述诊断：阅读真正执行的计划，而不是编译后的 SQL；检查读取侧和写入侧；将每个症状追溯到根本原因——并不特定于那个整合模型。这是任何工程师都会在 Databricks 上对任何缓慢增量模型执行的一套流程。正是这套流程被打包成一项技能：一份 AI 代理可按需执行的行动手册，从而使诊断能力随模型数量的增长而扩展，而不是取决于有多少工程师记得怎么做。

该技能逐步复刻了这个实例。它从查询历史中提取实际的语句族，而不是dbt compile输出，因为对增量模型来说，这些是不同的语句。它会读取一次运行的两侧：扫描侧指标（被裁剪的文件、读取的行数、溢写）和写入侧指标（写入行数与删除行数），因为放大效应只会在写入侧显现。然后它检查在整合模型中发现的同样三类故障：从不缩小的变更集（上游时间戳被重新生成，而不是一路传递下来）、无界的按资产重新计算（没有回看限制的窗口），以及浪费的工作（计算了但下游从未读取的列或窗口遍历）。每项检查都基于某个指标或计划信号，而不是凭直觉。

输出是一份报告，而不是静默修复：每项发现都连同其证据（扫描行数、溢写字节、计划节点）一起陈述，并配以建议的更改；在获得批准之前，不会对模型应用任何更改。一旦获批，用于证明该修复合理性的同一组前后指标会在下一次运行时重新测量，因此该技能会闭环验证，而不是假定修复已经生效。

收益在于一致性，而非新颖性。整合模型运行时背后的三个原因都很普通，在高负载下很容易被忽略（一个被重新生成的时间戳、一个无界窗口、一些死列）。运行一项技能来检测它们，重复执行的代价为零，并且它会在下一个模型上发现同类问题，避免其变成需要有人上报的 60% 运行时问题。

在 Databricks Jobs 上运行大型 dbt 项目

Databricks Workflows（Lakeflow Jobs）将 dbt 视为一等任务类型：dbt 项目可以与数据摄取和下游步骤一起，在一个受治理的工作流中进行调度、运行和监控，并共享重试和告警机制。最简单的版本是将整个项目作为单个 dbt 任务运行。这样做可行，但它是一个黑盒：一个模型失败，整个作业就会失败，无法查看、重新运行或监控单个模型。在这个规模下，这是一种运营风险。

解决方案是将 dbt 图作为单独的 Databricks 任务运行，每个模型、测试、种子和快照各一个任务。IFCO 使用databricks-dbt-factory来生成该图，这是一个独立的开源库（MIT 许可，位于GitHub和PyPI上）。它读取 dbt manifest 和一个作业模板，生成一个 Databricks Asset Bundle 作业，每个节点对应一个任务。只有当每个任务的启动成本很低时，按任务粒度拆分才有价值，这归结为三个机制：

• Notebook 任务类型。一个小的共享运行器 notebook 通过其 Python API 触发 dbt。dbt 生成 SQL 并将其发送到 SQL 仓库，因此 notebook 计算不处理任何数据。
• 基础环境。基础环境是预先构建的无服务器 Python 环境快照。这个环境已经包含dbt-databricks，因此每个任务从它启动，省去了全新任务本来需要支付的 pip install 成本。
• Manifest 注入。预先构建的 manifest 直接交给 dbt，从而跳过解析，并且每个任务将产物写入私有的本地目录。在一个由许多任务同时运行的大型项目中，这就是重新读取项目数分钟与做有用工作之间的差别。

在保持较小开销的情况下，扇出为运营提供了所需的能力：任务级可见性、仅针对失败模型及其依赖项的有针对性重跑、按模型日志、告警和测试，以及一个可以扩展的运行器（加载密钥、用 git SHA 标记运行，或几行代码发布到 Slack）。一切通过路径感知的 GitHub Actions 矩阵以 Databricks Asset Bundles 部署，按模型的运行时和成本从查询标签和系统表跟踪到一个带告警的仪表板，因此回归会在一天内显现，而不是在月度账单中。

大规模测试与质量

如果效率悄悄破坏了数字，那效率就毫无价值，因此质量是被强制执行的，而不是寄希望于它。每个模型都带有所有者和唯一性测试。主键以错误严重级别测试唯一性和非空。具有真实逻辑的模型（窗口函数、多个连接、非平凡宏）需要单元测试。dbt-bouncer阻止违反这些规则的提交，同时还有sqlfluff在 Databricks 方言上运行，并且契约在外部消费者读取的层上强制执行。

同样的纪律适用于测试本身的成本。对视图的检查会被物化或合并到更少的遍历中，因为基于视图的检查每次运行都会重新计算视图——基本检查变为列约束，测试的范围限定为增量数据。在本地，开发者延迟到生产 manifest，因此只有更改的模型会构建，而上游从生产读取。在 CI 中，单元测试和抽样数据测试在合并前对更改的模型运行。

基准测试

下一步是什么

如今的管道是批处理：数据摄取每天落地一次，语义层在其上运行，发出早期估计并在迟到数据到达时收敛。在合作期间推荐的三项工作将把它推进得更远，并为近实时 KPI 打开大门。

• 只消费变化的内容，使用 Change Data Feed。Delta 的 Change Data Feed 让下游模型只读取上游变化的行，而不是重新扫描其输入。应用于合并模型及其依赖项，它将协调工作从“扫描最近窗口”缩小到“处理确切移动的行”，这是在限制重新计算之后的自然下一步。
• 更低延迟的摄取。每天一次的加载可以转移到流式摄取路径，Auto Loader 或低延迟 Lakeflow Connect 连接器，而无需重新架构下游的任何内容。仅此一项就将新鲜度从一天缩短到几分钟。
• 声明式流式转换。相同的转换逻辑可以作为 Lakeflow Declarative Pipeline 持续运行，读取流，而不是作为夜间批处理，有状态处理在事件到达时处理每个资产的协调。

关键问题是业务需求，而不是技术。如果一个指标确实需要在几分钟内保持新鲜，而不是到第二天早上，这条路径在相同的受治理表上，用相同的 dbt 定义逻辑来交付它。如果每日刷新足够，批处理管道已经是更便宜的答案。

结论

解决方案的形态是清晰的劳动分工。转换逻辑保留在 dbt 中，模块化且经过测试，而数据以开放的 Delta Lake 格式保留在一个 Unity Catalog 治理模型下，因此血缘和访问控制在每次表重建后仍然有效，没有任何东西绑定到单一查询引擎。该 dbt 逻辑编译为为规模而构建的 Databricks 功能：liquid clustering、Delta 增量写入、动态文件修剪和 H3 地理空间函数。从真实查询计划而不是编译后的 SQL 进行诊断，减少到达昂贵排序和洗牌的行数，并将项目作为按模型的任务图运行，使运营获得可见性和低开销的安全重跑。最大的收益不是来自更大的集群，而是来自做更少的工作：触及更少的行、重新计算更少的资产，以及更少地重建表。