Spring AI与ELT管道融合：Java技术栈的智能数据清洗实践

1. 先弄清楚Spring AI 和 ELT 是怎么凑到一起的1.1 先说个明确结论这不是拿 AI 把 ETL 整个替换掉很多做数据工程的同事一听到“Spring AI ELT”第一反应是“用大模型替代 Data Pipeline”这个理解其实有偏差。我在实际项目里摸索下来的结论是Spring AI 在 ELT 流程里扮演的是“高难度 Transform 算子”和“管道智能调度助手”而不是替你重写整条数据链路。先解释一下 ELT 和 ETL 的区别因为很多人会把这两个概念混着说。ETL 是 Extract-Transform-Load 的顺序数据先在小范围或中间库里完成清洗加工再装载到目标存储。ELT 则是 Extract-Load-Transform先把原始数据完整无脑地 Load 进数据湖或数仓的原始层再在存储侧做大规模转换。ELT 流行的原因很现实数据量变大以后中间态加工的成本太高不如先落盘再用集群算力慢慢洗。那么 AI 在这里有什么价值原始数据落盘之后最脏、最需要“语义理解”的 Transform 环节往往是规则引擎和 SQL 洗不干净的。比如商品评论自动分类、非结构化地址清洗、客户信息归并、日志摘要生成这类任务你的 programmatic 规则写一万行也覆盖不全但用大模型来做效果一下子就上来了。而 Spring AI 恰好给了 Java 技术栈一个比较顺手的接入姿势能和现有 Spring Boot 项目无缝配合不至于为了一个清洗任务去单独搞一套 Python 服务。1.2 Spring AI 在 ELT 管道里的三个位置我在多次实践后把 Spring AI 放进 ELT 管道的位置归纳成三类也都对应不同的实现难度Extract 辅助层用模型解析上游数据源的 Schema、猜测字段含义、自动生成 API 拉取配置。这类任务不用很频繁适合离线跑。Transform 智能算子这是用得最多的场景比如消息分类、实体抽取、标准化格式、匹配归并、文本摘要。相当于把你原来容易脏、容易漏的规则逻辑替换成模型判断。Load 与质量校对层数据装载后用模型去校验结果字段是否完整、枚举值是否越界、敏感信息是否脱敏干净。这么做的好处是能兜底防止“垃圾数据进库”后污染下游报表。1.3 适合谁看不适合谁看这篇内容适合这样几类朋友第一类是 Java 后端开发想把大模型能力接到现有 Spring Boot 数据管道里第二类是数据工程/数仓开发被非结构化字段清洗折磨得够呛想找一个更聪明的方案第三类是技术负责人在做 AI 化数据平台的技术选型需要评估 Spring AI 的边界。不适合谁看呢如果你手里是 PB 级密集计算型管道每天跑几亿行纯数值加工那就别指望大模型来逐个解释行语义成本扛不住这是方向性问题。另外如果你完全没有 Java/Spring 基础想直接拿这个组合入门也比较吃力建议先熟悉一下 Spring Boot 的基本写法。2. 环境准备与工程骨架搭建2.1 技术栈选型与版本选择这部分我先给出一套实测下来比较省心的组合不是最新版本强迫症而是版本之间兼容性踩过坑之后沉淀下来的组件版本说明JDK17长期支持版本Spring Boot 3.x 必须Spring Boot3.2.x稳定版避免追 RC 版本Spring AI0.8.1目前 API 演进期版本锁定一个版本写代码更稳OpenAI SDKSpring AI 内置基于 REST 调用不绑定具体厂商实现Redis6.x/7.x做 Spring AI 的缓存与结构化输出缓存管理消息队列RabbitMQ任务流控控制管道水位PostgreSQL14存会话状态和 Pipeline 元数据注意Spring AI 目前版本迭代非常快0.8.1 是我建议锁定的版本。别一上来就用 snapshotAI SDK 接口变化快追新版本会让你大量时间花在适配调用签名上而不是解决业务问题。2.2 项目结构与依赖配置我一般会建一个多模块 Maven 工程因为 ELT 管道天然涉及多个解耦的环节。大致目录结构是这样的spring-ai-elt-demo/ ├── elt-common // 公共 DTO、工具类、缓存客户端 ├── elt-extractor // 数据抽取模块对接外部数据源 ├── elt-transform-ai // AI 智能算子模块Spring AI 核心逻辑 ├── elt-loader // 装载模块落库或同步到数仓 ├── elt-orchestrator // 管道编排模块 └── elt-web // 管理后台与 API 入口elt-transform-ai模块的 pom 关键依赖如下dependency groupIdorg.springframework.ai/groupId artifactIdspring-ai-openai-spring-boot-starter/artifactId version0.8.1/version /dependency dependency groupIdorg.springframework.ai/groupId artifactIdspring-ai-redis-spring-boot-starter/artifactId version0.8.1/version /dependency注意Spring AI 的 Redis starter 主要是做缓存存储和向量存储用的。如果只是简单调模型不一定要引它但 ELT 场景里大量重复文本会反复过模型缓存这一层几乎是必须的。2.3 基础配置与连接管理在application.yml里给出最小化配置spring: ai: openai: api-key: ${AI_API_KEY} base-url: ${AI_BASE_URL} chat: options: model: gpt-4o-mini temperature: 0.2 redis: host: localhost port: 6379这里有两个容易踩坑的点第一temperature在数据清洗场景下尽量不要超过 0.3否则模型输出不稳定同样的输入每次跑出来结果对不上数据管道最忌讳这个第二base-url可以做一层代理转发方便在模型服务切换时只改配置不动代码。我实际做的时候是在代理层做多模型的路由这样管道代码里只认 Spring AI 的ChatClient接口不用关心后端是哪个模型。3. 核心实现ELT 管道中的智能算子3.1 先做一个统一的 AI 服务层封装不要直接在业务代码里到处 newChatClient。我写了一个AiOperatorService把所有模型调用收敛到同一个类里这样方便统一加缓存、统一记录 token 消耗、统一处理异常。核心代码大致是这样的Service RequiredArgsConstructor public class AiOperatorService { private final ChatClient chatClient; private final StringRedisTemplate redisTemplate; private static final String CACHE_KEY_PREFIX ai:op:; public String complete(String prompt, AiOpOption option) { String cacheKey CACHE_KEY_PREFIX option.opName() : digest(prompt); String cached redisTemplate.opsForValue().get(cacheKey); if (cached ! null) { return cached; } ChatResponse response chatClient.call(new Prompt(prompt, option.toOptions())); String content response.getResult().getOutput().getContent(); redisTemplate.opsForValue().set(cacheKey, content, Duration.ofHours(option.cacheTtlHours())); return content; } private String digest(String text) { return DigestUtils.sha256Hex(text); } }为什么缓存要放在这一层而不是交给 Redis Starter 的自动缓存因为 ELT 管道的文本可能很长默认缓存机制对长文本 Key 不友好而且我要针对不同的算子设置不同的缓存有效期。拿地址清洗来说同一批收货地址可能在几天内重复出现缓存 48 小时就够了但新闻文本摘要这种场景缓存时间可以长一点。3.2 任务一无规则商品评论分类入宽表这个场景来自一个真实需求某电商平台有大量用户评论运营想按“质量投诉、物流投诉、售后问题、商品好评、咨询”等维度分类并进入宽表支持后续分析。原来的方案是维护一版关键词规则结果被“物流太慢了但质量很好”这种句子直接搞崩关键词命中还会互相打架。用 Spring AI 来改写这个算子的核心逻辑public class ReviewClassifierTransformer implements RowTransformerReviewRecord, ReviewClassified { private static final String SYSTEM_PROMPT 你是电商评论分类助手。请根据用户评论内容从以下类别中选择最合适的一个 - QUALITY: 关于商品质量、使用体验的问题或好评 - LOGISTICS: 关于物流速度、配送服务的反馈 - AFTER_SALE: 关于退换货、售后服务的反馈 - CONSULT: 咨询商品信息但未提及具体问题的内容 - POSITIVE: 整体正面评价但无明显指向 只输出类别英文代码不要输出其他解释。 ; public ReviewClassified transform(ReviewRecord record) { String userPrompt String.format(评论内容%s\\n评论时间%s, record.content(), record.time()); String category aiOperator.complete(SYSTEM_PROMPT, userPrompt, opOption(review_classify)); return new ReviewClassified(record.id(), category, record.content()); } }这里有个很关键的设计点系统提示词里严格限定了“只输出类别代码”并且类别枚举是封闭的。实际跑下来模型偶尔还是会输出别的词或者加一句解释。所以我在代码里加了一个兜底方法把模型返回的字符串做一次枚举映射映射失败就落到UNKNOWN分类再由下游补一个规则兜底。刚开始做的时候我没加这个兜底结果分类字段里混进了一些自由文本宽表直接没法做聚合。3.3 任务二地址清洗与标准化地址清洗是另一个高频脏活。真实世界的用户收货地址五花八门有“XX路XX号旁边那个奶茶店三楼”这种模糊地址也有“广东省深圳市南山区科技园路1号”这种相对标准的。规则引擎对前者基本无能为力但对大模型来说把模糊表达转成结构化字段是可以做到的。我设计的提示词是让模型输出 JSON再用 Jackson 反序列化成对象public class AddressStandardTransformer implements RowTransformerAddressRaw, AddressStandard { private static final String SYSTEM_PROMPT 你是一个中文地址解析器。请将用户输入的收货地址解析为以下 JSON 结构 { province: 省份, city: 城市, district: 区/县, detail: 详细地址, confidence: 0-1之间的小数 } 不存在的字段填空字符串。只输出 JSON不要输出其他内容。 ; public AddressStandard transform(AddressRaw raw) { String json aiOperator.complete(SYSTEM_PROMPT, raw.address(), opOption(address_std)); return objectMapper.readValue(json, AddressStandard.class); } }用 Structured Output 或者 JSON Schema 约束是更稳的方案但我在实践中发现如果模型版本不稳定强约束反而容易报错。退一步用“提示词限定 JSON Jackson 反序列化 失败重试一次”的做法整体通过率能到 97% 以上。剩下的 3% 基本是地址本身缺失太严重比如只有一个手机号这种模型也没办法直接进人工稽核队列。还有一个细节置信度字段非常重要。我建议对confidence 0.6的地址不要直接入库而是标记为待人工确认否则会把错误信息无声无息地洗进主数据里后面业务用起来全是坑。3.4 任务三客户脱敏与归并匹配脱敏这个任务看起来简单但场景一复杂就麻烦。比如用户的文本里既有“手机号 138xxxx1234”这种规整形式也有“电话13812345678微信号同号”这种粘连文本。用正则只能处理规整格式但模型可以顺带把微信号、邮箱、QQ 号一起找出来。我的做法是把脱敏分成两步第一步实体抽取第二步按类型替换。Spring AI 在这里就是做第一步的public class SensitiveInfoExtractor { private static final String SYSTEM_PROMPT 请从文本中找出所有敏感信息并返回 JSON 数组每个元素包含 { type: PHONE/EMAIL/ID_CARD/WECHAT/QQ, value: 原始值, start: 起始位置, end: 结束位置 } 如果没有敏感信息返回空数组。 ; public ListSensitiveEntity extract(String text) { String json aiOperator.complete(SYSTEM_PROMPT, text, opOption(sensitive_extract)); // 反序列化 位置校验 } }归并匹配也是 ELT 里的常见痛点同一个人在不同系统里留下的信息可能一条有手机号、一条有邮箱怎么判断是同一个人纯规则做不了模糊归并但模型可以。我把两条记录合并成一段文本让模型判断是否为同一实体并给出置信度。这个做法在小样本测试里效果不错但大数据量下要注意调用量控制不能一股脑全给模型判断否则成本会失控。实际做法是先做粗筛比如相同城市相似昵称把候选集压到 Top 5再让模型精细判断这样调用量能降一个数量级。3.5 任务四AI 生成摘要与汇总入库最后一个常见场景是把冗长的日志、工单、合同片段改写成结构化摘要落进汇总表里。对于帮助文档类内容摘要可以直接作为下游知识库的检索分片。public class LogSummaryTransformer implements RowTransformerLogRecord, SummaryRecord { private static final String SYSTEM_PROMPT 你是系统日志分析师。请阅读日志内容输出以下 JSON 结构 { errorType: 异常类型如 NETWORK_TIMEOUT/DB_CONNECTION_LOST/BIZ_RULE_FAIL/UNKNOWN, summary: 不超过50字的中文摘要, keyWords: [关键词数组最多5个], impactLevel: HIGH/MEDIUM/LOW } 只输出 JSON。 ; public SummaryRecord transform(LogRecord log) { // 组装消息时间、服务名、日志级别、正文 } }这个算子在生产环境里帮我们解决了一个很实际的痛监控系统的告警日志每天几千条工程师没精力逐条看有了摘要和关键词之后可以自动按 impactLevel 聚合把最高优先级的异常推到值班群。它不替代监控平台而是让监控结果消费起来更高效。4. 管道编排与缓存策略4.1 从单条调用到批量管道的编排方式在 Web 接口里单独调用一次 AI 算子不难但 ELT 管道动辄一天处理几万条数据这里面的编排逻辑跟单次调用完全是两码事。我需要先讲清楚 Pipeline 的状态机设计因为这是最容易被忽视的部分。我设计的核心抽象是ProcessorNode接口每个算子实现void process(DataContext context)。DataContext 维护当前行的状态包括原始数据、当前结果、错误信息、重试次数。管道编排器则负责逐个节点执行并记录每个节点的执行元数据public interface ProcessorNode { String nodeName(); void process(DataContext context); }实际执行时每一步都记录开始时间、结束时间、输入字节数、输出字节数、状态、错误信息。这些元数据刷到elt_node_execution_log表里便于排查“哪一步把数据搞坏了”。我见过很多项目省略这一步出了数据质量问题时只能靠猜——到底是在 Extract 阶段丢的字段还是在 Transform 阶段被洗错了完全说不清楚。4.2 Redis 缓存避免重复调用模型的三种用法AI 调用贵且慢ELT 场景天然会有大批重复数据缓存是省钱的第一道防线。我在实践里总结出三种缓存姿势Prompt 哈希缓存上文AiOperatorService里用的就是这种。优点是无脑通用缺点是长文本哈希和存储成本高建议只对短文本开。字段级缓存比如地址清洗只要省份/城市/区县相同且详址相似就能复用解析结果。这种需要自己设计相似度算法但效果比全文哈希好。结果表缓存把“输入文本 - 模型输出”的映射持久化到数据库带生效时间和失效逻辑适合低变更但高复用的主数据。用字段级缓存举个例子。我们对“广东省深圳市南山区科技园路1号”和“广东深圳南山区科技园路一号”这两个地址做了归一化处理统一把“一号”转成“1号”再去查缓存命中率提升了接近一倍。如果不做这种预处理模型会为两个基本一样的地址重复买单。提示如果做了缓存就一定得考虑失效逻辑。业务上地址会变分类标准会调整缓存过期时间我一般给 24-48 小时不要给长期缓存。特别是支付风控相关的判断更不能长期缓存。4.3 失败重试与水位控制ELT 管道跑批时最怕的不是模型返回慢而是某个算子在处理一批数据时集体超时或者上游接口抖动导致大量失败。我的策略分三层第一层是单条重试。对网络类异常超时、5xx用 Spring Retry 做最多 2 次重试间隔指数退避。注意不要对大模型返回质量类异常做重试比如模型输出格式不对重试 10 次也是错的应该走降级逻辑。第二层是队列水位控制。我把 ELT 任务投递到 RabbitMQ消费者按批拉取用信号量控制并发度。为什么这么做因为如果并发拉太高模型服务端会限流之后所有请求一起来又一起失败场面非常难看。我压测试出来的结论是单实例并发 8 比较稳再往上走模型侧 RT 会明显恶化。第三层是死信队列。重试耗尽后进入死信队列由人工或者定时任务重新触发。死信队列里的消息不丢弃也不自动无限重推比如有些深度学习模型对特定格式的调用会一直失败你推到天亮也是一样的结果。5. 测试与联调AI 能力的非确定性有多难搞5.1 智能检查模型校验 Transform 步骤的 SLAELT 管道上线之后怎么知道你的 Transform 算子工作正常传统方式是写一堆断言比如“字段不为空”“格式符合正则”。但 AI 算子输出是非确定的好的时候分类很准差的时候一批数据全部掉进 UNKNOWN。这时候我在实践里用了一个办法再用一个模型来当质检员。做法很简单抽样 N 条管道输出让质检模型判断“原始输入和模型输出是否语义一致”输出 pass/fail 评分低于阈值就触发告警。这相当于在管道末尾加了一道 AI 巡检。虽然多消耗一些 token但避免了“管道跑了一周下游报表全是错误分类数据”这种灾难。这套逻辑的实现不能直接用同一个模型最好用不同模型或至少不同 Prompt 风格。因为同一个模型既干活又自查容易“自己骗自己”比如它认定某条评论是物流类自查时也会倾向于认为这个分类没问题。换一个不同目标函数的模型来做独立性更强检测效果也更好。5.2 角色扮演验证模拟上游系统抖动写完管道联调阶段最需要模拟的是上游系统抖动场景。我在项目里写了几个模拟器JitteryApiSimulator负责随机延迟和随机 5xxSchemaChangeSimulator负责随机增删字段GarbageDataSimulator负责注入乱码、空值、超长字段。为什么这样做因为 ELT 管道在真实环境里的死法根本不是你精心准备的那些干净测试数据。等上游一次模型升级把某个枚举值从QUALITY换成QUALITY_ISSUE你的管道就大量落 UNKNOWN 了。提前用模拟器做混沌测试可以在上线前发现绝大多数这种兼容性问题。我印象最深的一个 bug模拟器注入了 50MB 的超长文本管道的字段校验没做长度上限直接把下游数据库的字段空间打满了。从那以后所有进入 Load 阶段的数据都强制做字段长度裁剪超过阈值先行截断再入数仓。这个教训让我意识到在 AI 管道的入口必须有硬性防线不能依赖模型自己掌握分寸。5.3 回归测试与向量召回对比AI 算子迭代了一版 Prompt 之后怎么证明新版本不会把旧版本好的能力丢掉传统回归测试只能测固定输入固定输出但 AI 的输出没法逐字对比。我采用的方案是向量召回对比对每批测试数据用 Embedding 模型把新老版本的输出都转成向量计算两版输出的余弦相似度低于阈值的数据点人工查看差异原因这个做法能很好地定位“语义回归”比单纯依赖文本 diff 靠谱得多。文本 diff 会告诉你“分类从 POSITIVE 变成了 CONSULT”但没法告诉你这个变化是合理的还是错误的。向量召回虽然也不能直接下结论但能缩小人工复核的范围把排查效率提升了一个量级。6. 参数调优与高频故障排查6.1 温度、Token、超时与重试参数怎么给很多同学第一次接 Spring AI 时对参数没有体感这里我给出一个经过多项目验证的参数基线参数取值说明temperature0.2结构化输出场景尽量的低保持确定性maxTokens500大多数 ELT 结构化输出不需要很长文本timeout30s-60s模型响应超时取大一点避免误判maxRetries2只对网络异常重试不对质量异常重试concurrency8单实例并发上限过高会被模型侧限流关于maxTokens我吃过亏。刚开始给模型设了 2048 的输出上限模型会啰啰嗦嗦输出一大堆解释后来把结构化任务统一压到 500模型输出反而更稳定。原因很简单输出长度限制收敛了模型的发散空间它只能输出短小精致的结果不会“自由发挥”。另一个常被人忽略的参数是frequencyPenalty和presencePenalty。ELT 场景建议两个都设为 0因为这些参数影响的是生成文本的丰富度而结构化输出我们希望它越“机械”越好。如果填了惩罚系数模型可能为了换词而改变分类表达输出反而不可控。6.2 高频问题速查表我在几个项目的联调和生产阶段收集了一些典型故障整理成速查表现象可能原因解决方法模型返回 JSON 解析失败提示词约束不强或模型被长文本干扰用 JSON Schema 约束增加格式校验后重试一次大批量数据分类落 UNKNOWN枚举值与模型惯用表达不一致检查枚举定义扩充分类映射在提示词里给出各分类的典型例句同一条数据每次跑结果不同temperature 过高未开缓存调低 temperature用缓存锁定结果管道重跑时不要重新生成调用频繁触发限流不够用并发太高或没有做缓存降低并发数加上 Redis/Prompt 缓存地址清洗结果省份城市对不上模型把模糊地址“脑补”完整了增加置信度阈值低置信度走人工队列报错上下文长度超限输入文本太长超出模型窗口在入口做截断对超长文本先做分块摘要这些故障里最隐蔽的是“地址清洗脑补”这个。模型会把模糊的信息补成看似合理的完整地址比如用户只写了“科技园路1号”模型硬是补出了“广东省深圳市南山区”。如果你不设置信度阈值这种数据会直接污染主数据。类似的问题在客户归并、实体抽取场景也常见AI 输出必须始终保留置信度或来源标记这是工程化落地的一条底线。6.3 成本与性能的实战控制手段最后说成本因为这是 AI 管道上线绕不开的话题。我控制成本的顺序从便宜到贵排序如下缓存优先。同一个 Prompt 只跑一次模型这是最便宜的成本优化手段一定要优先做。小模型优先。能跑通业务就不上大模型复杂的分类、抽取任务可以用 mini 级别模型先跑质量不达标再升级。实测中地址清洗用大模型还是 mini 模型差别不大但分类任务差异明显。粗筛后再过模型。先用便宜的规则或向量方法做初筛把简单明确的数据分流掉只把模糊的数据交给大模型判断。这样模型调用量常常能下降 50%-70%。批量提示词压缩。把 10 条短文本拼进同一个 Prompt让模型一次性输出 10 行结果可以显著降低单条处理的 Token 开销。但注意批量过大会降低格式稳定性我控制在 5-10 条一批。性能优化跟成本优化是一体两面。调用模型前先做文本压缩去掉换行符、全角转半角、去除无意义语气词能有效减少 token。跑批时按数据源分片每个分片独立控制并发避免一个慢任务拖垮整个管道。写在最后的几点体验这套“Spring AI ELT”的组合我在实际项目里前前后后跑了快半年最大的体会是它真正的价值不在于让某个单点任务变得更聪明而在于把一批过去靠规则引擎写不明白、靠人工处理又太慢的语义类脏活变成了一种可编码、可复用、可观测的工程能力。分类算子、地址清洗、脱敏、摘要这些能力沉淀成公共模块之后新管道接入新数据源的边际成本明显降低。如果你正准备把这套方案落地到自己的团队我的建议是先别急着做大而全的 AI 平台找一个最痛的非结构化清洗场景把一个算子做好做透。把这个算子的缓存链路、失败处理、质量校验全部跑通挨个检查“当模型给错答案时管道怎么办”再去横向复制到其他任务。AI 管道的技术难度不在于怎么调用模型而在于怎么设计一个能让非确定性输出乖乖待在确定性流程里的容器。这个容器搭好之后整个数据团队都会开始重新看待那些过去只能靠人力堆的老大难任务。