★ FEATURED ARTICLE 1. 为什么我在 ELT 流水线里塞进一个大模型我第一次把大模型放进 ELT 流水线其实是被一列地址数据逼的。那段时间我接了一个订单类数据同步场景上游好几个系统往同一个目标表里灌数据表面上都是“收货地址”实际上玩法完全不一样。有的系统规规矩矩按省市区字段拆好了有的系统只留了一个大文本框里面还混着联系人姓名、电话、甚至是备注“放前台”。这种数据进 ELT 之后传统手段是正则加映射字典。正则看起来能兜住大部分情况但真实业务数据最擅长的就是用各种姿势突破正则的假设。同一个“科技园路”在一个系统里叫“science park road”在另一个系统里叫“高新中一道”人工维护一张互相映射的字典表确实能撑一阵但撑不了多久因为数据里永远有字典没记录过的写法。那段时间我的感受是ELT 的 T 字头真正难的不是搬数而是把搬到一半的数据转变形。原来的做法是“你写死规则数据来适配规则”只要上游业务一改表结构、一改下拉菜单规则就废了。而大模型在“理解模糊输入并转化成标准结构”这件事上天然比正则字典更接近人类处理数据的方式它不用见过所有写法只要能读懂语义就能按预设格式吐出标准字段。但把这个想法落到工程里绕不开一个现实我所在的数据集成团队长期是 Java/Spring 技术栈整条数据链路、调度任务、监控报警都是 Spring 家族的。让我因为一个小需求去另搭一套 Python 服务维护成本太高跨语言排查也麻烦。Spring AI 的出现在当时正好给了我们一个更顺手的选项。它本质上就是把对大模型的调用抽象成 Spring 风格的一组能力比如 ChatClient、消息模板、结构化输出、向量模型封装让我可以在已有的 Spring Boot 工程里直接用类似写 Mapper 的感觉去接大模型。更关键的是Spring AI 把“提示词”和“数据转换逻辑”分开了。基于大模型的 ELT 转换最终会演变成一个需要持续维护的模块如果没有工程化封装团队里很快会变成“一段长字符串写在代码里谁都不敢动”。Spring AI 的 PromptTemplate 和输出转换器能让这段东西看起来更像正常代码这是我在选型时非常在意的一点。当然必须先说清楚边界AI 在 ELT 里的角色不是替代已有的数据引擎。你不能让它去跑几千万行的聚合也不能让它去做强一致性的事务操作。它应该只出现在那些“规则写不死、正则扛不住、字典永远差一页”的环节里比如地址清洗、名称归一、分类映射、质量评分这些偏语义处理的子步骤。前面是 SQL、Spark、Flink 的地盘后面才是大模型发挥价值的区域。理解这个边界后续才不会把架构搞得越来越贵、越来越玄。2. 整条流水线的落点设计先把转换拆成能 AI 把关的步骤说实话我一开始犯过一个小错想直接把“读原始表 - 让大模型清洗 - 写目标表”这样一个两端简单的流程跑通。后来发现这种粗暴设计的问题太多了最痛的一点是你没法区分哪些记录真的需要 AI哪些只是过来凑热闹的。如果每条数据都调一次大模型成本和响应时间都会迅速失控。所以在正式实现之前我花了不少时间重新切分层级。当前稳定下来的结构大概是这样的源数据落地区 - 预过滤层 - AI 转换区 - 程序校验层 - 目标数仓层源数据落地区只做一件事就是把上游数据原样保存下来不做任何改动。预过滤层用普通速度和规则把这些数据过一遍能走常规映射的走常规映射只有“规则无法确定”或“字段格式异常”的记录才会继续往后走。AI 转换区里面放着一个个由 Spring AI 封装好的转换算子它们按照数据语义去归一化、格式化、补全。程序校验层则负责把 AI 的输出再检查一遍字段缺失就回退默认值枚举不合法就拒绝写入。最后只有完整校验通过的数据才进入目标层。这个设计的核心思路是AI 不是整条管道的主角而是特殊数据流的一个旁路处理器。绝大多数数据用便宜、高速的传统方式处理只有小部分“人类都拿不准”的数据才交给模型判断。这不只是为了省钱更是为了让整条 ELT 流程保持确定性。数据仓库的下游通常关心的是“结果可复现”如果每次跑同步任务同一条脏数据因为大模型温度参数抖动得到不同结果下游做数据对账会非常痛苦。另一个容易被忽略的点是触发条件。我最终把 AI 转换变成了一个“按需触发”的算子而不是流水线上的固定步骤。触发条件通常包括字段校验失败、枚举映射找不到、文本长度或特征值超出已知范围、或者持续一段时间内同步异常率偏高。这样做的好处是当上游数据质量变好时模型调用会自动减少成本跟着降当上游又“创新”出新的脏数据格式时算子会自动兜住新情况。整个过程不需要人肉去改正则。落到 Spring 生态里我把它做成了一个独立服务内部封装了完整的 AI 转换能力对外只暴露非常简单的接口public interface NormalizePort { NormalizeResultT normalize(T raw); boolean support(T raw); }这样 ELT 管道里的调度、批处理、流处理组件都不需要关心大模型的具体实现。它们只负责拉数据、调用端口、收结果。真到了某一天想换模型服务商或者想从云端模型切到本地推理只需要改 Spring AI 的接入配置和这个算子内部实现不会牵动整条管道。这种分层还有一个隐性好处每一条调用大模型转换的数据都会留下原始样本和转换结果对应的记录。它既是可审计日志也是以后做回归测试的样本集。大模型转换这类东西最难的不是写出来而是改完了不知道会不会把以前能处理好的样本弄坏。有了这种“原始值 目标值”的留痕我后来在做提示词调整时都是直接在历史样本集上跑一遍对比。3. 用 Spring AI 写一个“能读懂脏数据”的转换算子直接上案例吧。我用得最多的场景是“电话地址联系人信息”从一个大文本框里拆出来。之前用正则写了好几版每次都折在“没有空格、没有分隔符、备注又特别长”的样本上。后面换成 Spring AI 的算子逻辑就清爽很多。先看依赖。在 Spring Boot 工程里引入 Spring AI 的 BOM再根据你实际接的大模型服务选择对应 starter。我这里不写死具体服务商的名字因为不同厂商的 Starter 包名略有区别但套路一致dependencyManagement dependencies dependency groupIdorg.springframework.ai/groupId artifactIdspring-ai-bom/artifactId version1.0.0/version typepom/type scopeimport/scope /dependency /dependencies /dependencyManagement实际使用的时候我更推荐只注入 ChatClient 而不是直接依赖底层大模型对象。Spring AI 的 ChatClient 已经把对话消息、提示词模板、结构化输出这些基础能力封装好了日常开发根本不碰模型通信细节Service public class AddressSemanticExtractor { private final ChatClient chatClient; public AddressSemanticExtractor(ChatClient.Builder builder) { this.chatClient builder.build(); } public ExtractResult extract(String rawText) { BeanOutputConverterExtractResult converter new BeanOutputConverter(ExtractResult.class); PromptTemplate promptTemplate new PromptTemplate( 你是一名数据清洗工程师。请从下面这条原始文本中提取信息。 规则 1. 只提取文本中真实存在的信息禁止猜测或补全。 2. 如果某个字段无法确定填空字符串。 3. 将结果输出为 JSON字段结构如下 {format} 原始文本 {rawText} , Map.of( rawText, rawText, format, converter.getFormat() )); String response chatClient.prompt() .system(你只负责数据提取与标准化输出不回答任何其他问题。) .user(promptTemplate.render()) .call() .content(); return converter.convert(response); } }对应的输出结构是这样一组普通 Java 类public record ExtractResult( String province, String city, String district, String street, String contactName, String phone ) {}这段代码看着不多但几个地方值得细说。第一BeanOutputConverter是 Spring AI 里很实用的一个工具。它会把 Java 类型结构转成模型能理解的输出格式说明塞进提示词里再把模型返回的文本转换回 Java 对象。它解决的核心问题是“模型输出天然不稳定”你提示它按 JSON 输出它可能多给你补一段解释文字也可能把字段名从下划线变驼峰。结构化输出把这种不确定性尽量降到最小。第二温度参数要调低。处理 ELT 场景的数据我一般把 temperature 放在 0 到 0.2 之间。温度越低模型越倾向于选择稳定、保守的表达方式幻觉和随机性自然少一些。这个参数了解生成式模型的人都知道但在 ELT 管道里它重要到值得单独强调因为下游每个字段都会被直接落库一丁点随机都可能变成脏数据。第三提示词里明确要求“不确定就填空字符串”。很多人在调大模型做提取时默认模型会把缺失的信息自行脑补出来这在普通对话场景里没问题但在数据管道里是不可接受的。我们在提示词里反复强调“禁止补全”就是为了把模型从“乐于助人的聊天者”掰成“保守的数据录入员”。第四这只是一个算子不是流水线的终点。我拿到ExtractResult之后还会走一层程序校验比如 phone 字段如果不是合法的手机号就不采用模型给的结果而是回退到原始字符串里的首段比如 province 字段如果不在我们维护的省份枚举里就标记异常转人工。这个校验层是绝对必要的因为目前没有任何模型能在所有样本上都稳定输出程序校验是安全网。4. 上线之后真正要盯的三个问题超时、成本和幻觉把算子写进管道后我最初只关注转换效果觉得只要模型理解能力强一切问题都能解决。结果上线后连续出了好几次状况总结下来最影响稳定性的就是超时、成本和幻觉这三件事。这三件事不提前安排好AIELT 会从“智能数据清洗”变成“定时炸弹”。超时问题比想象中严重。大模型接口不是数据库连接它的响应时长波动非常大有时候几百毫秒就回包有时候一个小输入因为服务端负载高能拖到几十秒。我们的 ELT 任务通常有严格的调度窗口不能允许一个算子无限等待。所以我做两件事第一给外部调用设置连接超时和读取超时把单次调用的最长时间控制在比如 30 秒以内第二接入重试和降级机制超时以后先重试一次如果还不行就把这条记录原始值原样写进异常表同时不让整批任务卡死。这比“把所有问题都抛给大模型”要安全得多。成本问题也很好算算完你会立刻改变触发策略。假设一条原始记录平均有 800 个输入 token、模型回复 250 个输出 token如果十万条异常记录全部调用大模型就是 8000 万输入 token 加 2500 万输出 token。这个量级再乘上单位价格账单可不是小数字。而如果只是其中两万条“真正偏离规则”的记录触发调用成本直接缩到四分之一。所以我在最前面加了一道便宜的预过滤能用普通规则处理的绝不动用模型这条经验比任何模型调优都省钱。当然还可以做结果缓存。对 ELT 场景来说同样的脏数据很可能在多次同步任务里反复出现。我维护了一张“原始文本摘要到转换结果”的缓存表命中缓存就直接复用结果。只要缓存键设计得好比如基于原始文本的 hash 值这个机制能把重复调用的比例降得很明显。我处理过的一个业务里缓存命中率甚至到了三成以上因为上游数据源老数据被反复补推的情况非常多。幻觉问题则是这三件事里最隐蔽的。模型在提示词要求下依然可能生产出“看起来合理但实际不存在”的内容尤其是在名称、编号这类细节上。比如我让模型从历史备注里提取渠道来源它偶尔把“看起来像渠道名”的普通词挑了出来其实原文里根本没有这个渠道。对付这个问题我把温度调到最低同时在下游做了严格的枚举校验和关联校验只要目标字段不在受控字典里一律判定转换失败并返回原始值而不是强行接受模型输出。问题典型表现我在实践中采用的措施超时上游接口慢批次任务卡住连接超时限制、读取超时限制、重试一次、超时后写入异常表成本错误触发率高账单暴涨预过滤减少触发、结果缓存、压缩提示词、限制并发请求数幻觉字段被模型“脑补”出不应有的内容温度调到最低、输出结构约束、下游枚举校验、失败回退原始值还有一个容易忽略的实操细节在 ELT 管道里调用模型最好用同步调用而不是流式输出。流式响应适合面对真实用户逐字展示但在后台数据任务里流式不仅增加恢复生产的复杂度还会让超时控制变得很别扭。我是用call().content()这种同步方式拿完整结果宁可多等几百毫秒也要保证逻辑简单、结果可复制。并发度也要控制。很多团队一看到有十万条数据要处理立刻开几十个线程去并发调模型结果被限流或者触发服务端阈值反而更不稳定。我自己的做法是先测出模型服务的可用并发窗口再用一个可控的线程池把 TPS 压在峰值的七折左右。宁慢勿快因为 ELT 任务关心的是最终都能处理完而不是瞬间冲高然后被熔断。5. 再往前走AI 加 ELT 还能延伸到哪里算子和管道跑顺之后我开始主动找更多可以把 AI 塞进 ELT 的地方因为一旦治理逻辑成熟同样一套 Spring AI 能力可以复用到很多以前靠人肉维护的环节。最直接的一个是语义字段映射。传统 ELT 在做字段映射时通常靠配置表源字段名和目标字段名在界面上手动拖拽关联。可当源系统特别多、字段命名五花八门时配置表会膨胀到没人敢改。我用大模型做过一次“字段名翻译”把源系统字段名一次性丢给模型让它对照目标模型的标准字段输出映射结果人工确认后写回配置表。这不算完全自动化但已经把一次特别繁琐的梳理工作压缩到了一小时代以内。另一个值得投入的方向是主数据归一。很多企业级数据仓库都有一张“标准客户表”或者“标准物料表”合并来自多个系统的重复记录是 ELT 阶段非常耗时的工作。用模型做语义相似度判断和属性合并比单纯按名称精确去重要可靠得多。Spring AI 里也提供了向量模型相关的封装可以把文本转成嵌入向量然后交给向量索引做相似性召回再结合大模型做最终确认。还有就是数据质量评分。同步管道里的脏数据往往造成下游各种“奇难杂症”但真正发现时已经晚了。我在 ELT 的预过滤层上放过一个评分算子专门对每个字段做质量评分比如空值率、格式异常率、与历史值的偏离程度。大模型在这里的角色是解释性的它能够把一堆质量指标用自然语言总结成“该字段近三天突然出现大量英文地址疑似新渠道接入导致”让值班工程师一眼定位问题。这个能力不是模型不可替代但确实大幅降低了人肉看报表的成本。比较有意思的还有敏感数据识别。有些系统往数仓同步日志信息时会把身份证、手机号等数据不自觉地带进来。传统做法是维护一套正则规则去扫描但很多敏感信息不是固定格式。我试过让模型基于上下文去判断“这段文本是否包含个人信息”效果比纯正则覆盖面大很多漏报率也低不少。整个项目做到后面我对“Spring AI ELT”这个组合的定位已经很清楚它不是拿 AI 制造一个黑盒魔法管道而是把大量以前只能靠人肉经验和脆弱的规则实现的数据理解工作变成一个可以迭代、可以测试、可以治理的工程模块。现在回头看我最大的体会是别上来就想做“全自动 AI 数据管道”那不现实也容易失控。更稳妥的路子是从最脏的一列数据开始实现一个非常窄但完整的 AI 清洗算子配上校验和回退机制然后慢慢地把周围的问题一个接一个地收编进来。等把这些窄而可靠的算子串到一起时你拥有的才是一条真正可持续演进的新 ELT 流水线。
阅读完成 · 觉得有帮助？
