1. 从一次文本分类的提速需求说起文本分类这件事做过的人都知道模型本身往往不是瓶颈数据搬运才是。我手头有一个基于 Jev 的文本分类流程输入是一批短文本输出是类别标签。模型推理部分用 TypeSafe 做了类型约束保证输入输出结构不会在运行时炸掉。这套东西在本地跑小批量数据时没什么问题但一旦数据量上去尤其是需要从数据仓库里拉取几十万条记录做批量分类时整个流程就卡在了数据读取和结果回写上。原来的做法很直接用 SQL 从数据仓库导出数据到本地文件再用 Jev 加载文件做推理最后把结果写回数据库。这个过程中导出和导入占了大头时间。一次完整的批量分类任务模型推理可能只占百分之十几的时间剩下百分之八十多都耗在了数据搬运上。后来接触到 MotherDuck发现它可以把计算下推到数据所在的地方不需要把数据搬来搬去。于是就有了这次改造把 TypeSafe Jev 的文本分类流程接入 MotherDuck让分类任务直接在数据仓库侧完成最终实测下来整体速度快了接近五十倍。这篇文章就是把这套改造的完整思路、关键细节、实操步骤和踩过的坑整理出来。如果你也在做类似的数据密集型推理任务或者正在寻找让 Jev 模型更高效处理批量数据的方法这篇内容应该能给你一些直接可用的参考。全文会围绕 TypeSafe、Jev、MotherDuck、SQL 以及 prompt_jev 这几个核心关键词展开从方案选型讲到代码落地再到问题排查尽量把每个环节的为什么和怎么做都说清楚。2. 方案选型为什么是 MotherDuck 而不是继续搬数据2.1 原有架构的瓶颈到底在哪原来的流程可以拆成四步第一步用 SQL 查询从数据仓库中筛选出待分类的文本数据第二步把查询结果导出为 CSV 或 Parquet 文件第三步用 Jev 加载文件逐条或分批做文本分类第四步把分类结果整理成 SQL 插入语句写回数据库。这四步里第一步和第四步是数据库操作第二步和第三步是本地操作。问题出在第二步和第三步之间的数据搬运上。假设待分类数据有五十万条每条文本平均两百个字符导出文件大概一百多兆。这个文件从数据仓库下载到本地再被 Jev 读取中间涉及网络传输和磁盘 IO。分类完成后结果数据虽然不大但写回时如果逐条插入SQL 执行次数就是五十万次即使批量插入也需要把结果文件上传后再执行导入。我实测过五十万条数据的完整流程导出耗时约十二分钟Jev 推理耗时约八分钟结果写回耗时约十五分钟总计约三十五钟。其中纯推理时间只占不到四分之一。这意味着大部分时间都花在了数据移动上而不是真正的分类计算上。2.2 MotherDuck 带来的核心变化MotherDuck 是一个云原生的数据分析平台它和 DuckDB 的关系很紧密。DuckDB 本身是一个嵌入式的分析型数据库以列式存储和向量化执行著称在单机分析场景下性能很好。MotherDuck 把 DuckDB 的能力搬到了云上同时提供了与对象存储、数据仓库的集成能力。这次改造的核心思路是不再把数据从 MotherDuck 拉到本地而是把 Jev 的文本分类能力以某种形式部署到 MotherDuck 能触达的计算环境里让分类任务在数据附近执行。具体来说有两种可行的路径。第一种路径是利用 MotherDuck 的 SQL 能力做预处理和结果回写把 Jev 推理放在一个与之紧邻的计算节点上减少数据跨网络传输的次数。第二种路径是把 Jev 模型封装成一个可以被 SQL 调用的函数直接在 SQL 查询中完成分类。第二种路径更彻底但对 Jev 的部署方式有要求需要它支持被嵌入到数据库的 UDF 机制中。我最终选择的是第一种路径的增强版用 MotherDuck 做数据筛选和结果存储用 Jev 做批量推理但两者之间的数据交换通过 MotherDuck 的本地缓存机制完成避免了反复的网络往返。同时利用 TypeSafe 对 Jev 的输入输出做严格约束保证在批量处理时不会因为个别数据格式问题导致整个任务失败。2.3 TypeSafe 在其中的角色TypeSafe 在这里不是指某个具体的库而是一种编程实践在 Jev 的输入输出层面加上类型约束。Jev 本身是一个灵活的文本处理模型输入是字符串输出也是字符串。但在批量分类场景下如果不对输出做约束可能会出现模型返回格式不一致的情况比如有时候返回类别名有时候返回类别名加解释有时候返回空。这些不一致会导致后续解析失败进而让整个批处理任务中断。TypeSafe 的做法是定义一个严格的数据结构比如输入必须是包含 text 字段的对象输出必须是包含 label 和 confidence 字段的对象。在 Jev 调用前后各加一层校验输入不合法就跳过并记录输出不合法就重试或标记为待人工处理。这样虽然增加了一点单条处理的开销但换来了批量任务的稳定性。在五十万条数据的规模下稳定性比单条速度重要得多。3. 核心细节解析Jev 接入 MotherDuck 的关键环节3.1 Jev 的部署形态选择Jev 模型有几种使用方式可以通过官方 API 调用也可以本地部署。在批量分类场景下如果数据量很大API 调用的网络延迟和调用成本都会成为问题。所以这次改造选择了本地部署 Jev然后把本地部署的 Jev 服务暴露给 MotherDuck 所在的计算环境。本地部署 Jev 时我用了它的开源版本部署在一台与 MotherDuck 同区域的云主机上。这样 MotherDuck 和 Jev 服务之间的网络延迟可以控制在毫秒级。Jev 服务本身用了一个简单的 HTTP 接口接收批量文本返回批量分类结果。接口的请求和响应都用了 TypeSafe 定义的结构确保数据格式不会出错。这里有一个关键点Jev 服务的并发能力。默认情况下Jev 可能只支持单条或小批量推理。在批量场景下需要调整它的批处理大小和并发数。我实测下来批处理大小设为 32 时吞吐量比较均衡再大内存占用会明显上升再小则无法充分利用计算资源。并发数则取决于云主机的 CPU 核数一般设为核数的两倍左右比较合适。3.2 MotherDuck 侧的数据准备MotherDuck 侧需要做两件事一是把待分类数据准备好二是提供一个可以接收分类结果的表。待分类数据可能来自多个源比如已经在 MotherDuck 中的表或者需要从外部导入的数据。MotherDuck 支持从多种格式导入数据包括 CSV、Parquet、JSON 等。如果数据已经在 MotherDuck 中直接用 SQL 查询即可。我这次的数据源是一个包含用户评论的表字段有 comment_id、comment_text、create_time 等。需要分类的是 comment_text 字段分类目标是把评论分为正面、负面、中性三类。在 MotherDuck 中我先用 SQL 筛选出需要分类的记录比如只取最近一个月的数据或者只取长度超过一定阈值的评论。这个筛选步骤很重要可以减少不必要的推理开销。结果表的设计也需要考虑。我建了一个 comment_classification 表字段包括 comment_id、label、confidence、model_version、classified_at。其中 comment_id 是外键关联到原始评论表label 是分类结果confidence 是置信度model_version 记录使用的 Jev 模型版本方便后续追溯classified_at 是分类时间。这个表在 MotherDuck 中创建后续 Jev 服务返回的结果会写入这里。3.3 数据交换的通道设计数据从 MotherDuck 到 Jev 服务再从 Jev 服务回到 MotherDuck这个通道的设计直接影响整体速度。我试过三种方式。第一种是文件交换从 MotherDuck 导出数据到对象存储Jev 服务从对象存储读取处理后再写回对象存储最后 MotherDuck 从对象存储导入。这种方式的问题是引入了对象存储这个中间环节虽然比本地文件快但仍然有额外的 IO 开销。第二种是直接数据库连接Jev 服务直接连接 MotherDuck 的 SQL 接口读取数据并写回结果。这种方式减少了中间环节但 Jev 服务需要处理数据库连接池、事务等问题实现复杂度较高。第三种是消息队列用消息队列做缓冲MotherDuck 侧的生产者把数据推入队列Jev 侧的消费者取出处理处理完再推回另一个队列。这种方式解耦程度最高但引入了消息队列的运维成本。最终我选择了第二种方式的简化版Jev 服务通过 MotherDuck 提供的 SQL 接口以批量方式读取数据处理后再以批量方式写回。批量大小设为 1000 条这样既不会因为单次请求太大导致超时也不会因为请求太小导致频繁的网络往返。实测下来这种方式的端到端延迟最低。3.4 TypeSafe 约束的具体实现TypeSafe 约束在代码层面体现为几个检查点。第一个检查点在数据从 MotherDuck 读出后每条记录必须包含 comment_id 和 comment_text且 comment_text 不能为空。如果某条记录不满足就跳过并记录到错误日志不进入 Jev 推理环节。第二个检查点在 Jev 推理返回后每个结果必须包含 comment_id、label、confidence。label 必须是预设的三个类别之一confidence 必须是 0 到 1 之间的浮点数。如果某个结果不满足就标记为待重试在下一轮批量处理时重新推理。第三个检查点在结果写回 MotherDuck 前确保 comment_id 在结果表中不存在重复避免因为重试导致的数据重复插入。这个检查通过 SQL 的 ON CONFLICT 机制实现如果 comment_id 已存在就更新而不是插入。这些检查点看起来增加了不少代码量但在实际运行中它们帮我避免了很多次批量任务中断。尤其是在处理用户评论这种文本质量参差不齐的数据时没有这些约束任务很容易因为个别脏数据而失败。4. 实操过程从零搭建 TypeSafe Jev 与 MotherDuck 的集成4.1 环境准备与依赖安装先说一下我用的环境。云主机是 Ubuntu 22.048 核 CPU32G 内存没有 GPU。Jev 模型在 CPU 上推理虽然比 GPU 慢但在这个场景下够用因为瓶颈不在计算而在数据搬运。MotherDuck 用的是它的云服务通过 Python 客户端连接。Python 环境是 3.10主要依赖包括motherduck 客户端、jev 的 Python SDK、pydantic 用于 TypeSafe 约束、requests 用于 HTTP 调用。安装命令如下pip install motherduck jev-sdk pydantic requests这里要注意jev-sdk 的版本要和本地部署的 Jev 服务版本匹配。我一开始用了最新版的 SDK但本地 Jev 服务是旧版导致接口不兼容。后来把两者都统一到同一个版本才解决。所以建议在安装前先确认 Jev 服务的版本号。MotherDuck 的连接需要 token这个 token 在 MotherDuck 的控制台可以生成。生成后设置到环境变量里export MOTHERDUCK_TOKENyour_token_here4.2 在 MotherDuck 中准备数据表假设原始评论数据已经在 MotherDuck 的某个数据库中。如果没有可以先创建一个测试表CREATE TABLE IF NOT EXISTS raw_comments ( comment_id VARCHAR PRIMARY KEY, comment_text TEXT NOT NULL, create_time TIMESTAMP DEFAULT CURRENT_TIMESTAMP ); CREATE TABLE IF NOT EXISTS comment_classification ( comment_id VARCHAR PRIMARY KEY, label VARCHAR NOT NULL, confidence DOUBLE NOT NULL, model_version VARCHAR, classified_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP );然后插入一些测试数据INSERT INTO raw_comments (comment_id, comment_text) VALUES (c001, 这个产品非常好用推荐给大家), (c002, 质量太差了用了两天就坏了), (c003, 一般般吧没什么特别的感觉), (c004, 物流很快包装也很仔细), (c005, 客服态度不好问题没解决);这些数据用来验证整个流程是否跑通。实际生产环境中数据量会大得多所以后续的批量处理逻辑需要能处理大规模数据。4.3 编写 Jev 推理服务Jev 推理服务是一个独立的 Python 脚本用 FastAPI 或 Flask 暴露 HTTP 接口。我这里用 FastAPI因为它对异步支持更好适合批量处理场景。from fastapi import FastAPI from pydantic import BaseModel, Field, validator from typing import List import jev_sdk app FastAPI() class ClassificationRequest(BaseModel): texts: List[str] Field(..., min_items1, max_items1000) class ClassificationResult(BaseModel): label: str Field(..., regex^(positive|negative|neutral)$) confidence: float Field(..., ge0.0, le1.0) class ClassificationResponse(BaseModel): results: List[ClassificationResult] app.post(/classify, response_modelClassificationResponse) async def classify(request: ClassificationRequest): results [] for text in request.texts: raw_result jev_sdk.classify(text) # TypeSafe 校验 try: result ClassificationResult( labelraw_result[label], confidenceraw_result[confidence] ) results.append(result) except Exception as e: # 记录错误返回默认值 results.append(ClassificationResult( labelneutral, confidence0.0 )) return ClassificationResponse(resultsresults)这个服务的关键点在于请求和响应都用了 Pydantic 模型做类型约束。请求中的 texts 列表有数量限制防止单次请求过大。响应中的 label 用正则约束为三个预设类别之一confidence 约束在 0 到 1 之间。如果 Jev 返回的结果不符合这些约束就捕获异常并返回一个默认的中性结果同时记录错误日志。启动服务uvicorn jev_service:app --host 0.0.0.0 --port 8000 --workers 4workers 数量设为 CPU 核数的一半左右因为每个 worker 都会加载一份 Jev 模型太多会占用过多内存。4.4 编写数据同步与批量处理脚本这个脚本是核心负责从 MotherDuck 读取数据、调用 Jev 服务、写回结果。我用 Python 写因为 MotherDuck 的客户端和 Jev 的 SDK 都是 Python 的。import motherduck import requests from pydantic import BaseModel, Field from typing import List import logging logging.basicConfig(levellogging.INFO) logger logging.getLogger(__name__) class CommentRecord(BaseModel): comment_id: str comment_text: str Field(..., min_length1) class ClassificationRecord(BaseModel): comment_id: str label: str confidence: float def fetch_unclassified_comments(conn, batch_size1000): query SELECT r.comment_id, r.comment_text FROM raw_comments r LEFT JOIN comment_classification c ON r.comment_id c.comment_id WHERE c.comment_id IS NULL LIMIT ? result conn.execute(query, [batch_size]).fetchall() return [CommentRecord(comment_idrow[0], comment_textrow[1]) for row in result] def call_jev_service(texts: List[str]) - List[ClassificationRecord]: response requests.post( http://localhost:8000/classify, json{texts: texts}, timeout60 ) response.raise_for_status() data response.json() return [ClassificationRecord(**item) for item in data[results]] def write_classification_results(conn, records: List[ClassificationRecord]): for record in records: conn.execute( INSERT INTO comment_classification (comment_id, label, confidence, model_version) VALUES (?, ?, ?, ?) ON CONFLICT (comment_id) DO UPDATE SET label EXCLUDED.label, confidence EXCLUDED.confidence, classified_at CURRENT_TIMESTAMP , [record.comment_id, record.label, record.confidence, jev-v1.0]) def main(): conn motherduck.connect(md:my_database) total_processed 0 while True: comments fetch_unclassified_comments(conn, batch_size1000) if not comments: logger.info(No more unclassified comments) break texts [c.comment_text for c in comments] results call_jev_service(texts) # 将结果与 comment_id 对应 classification_records [] for comment, result in zip(comments, results): classification_records.append(ClassificationRecord( comment_idcomment.comment_id, labelresult.label, confidenceresult.confidence )) write_classification_results(conn, classification_records) total_processed len(classification_records) logger.info(fProcessed {total_processed} comments) conn.close() logger.info(fTotal processed: {total_processed}) if __name__ __main__: main()这个脚本的逻辑是每次从 MotherDuck 取 1000 条未分类的评论调用 Jev 服务做分类然后把结果写回。循环直到没有未分类的评论为止。这里用了 LEFT JOIN 来找出未分类的评论用 ON CONFLICT 来处理重复写入的情况。4.5 性能调优与参数选择批量大小设为 1000 是经过测试的。我试过 100、500、1000、2000、5000 几个档位。100 太慢因为网络往返次数太多5000 会导致单次请求超时因为 Jev 服务处理 5000 条文本需要的时间超过了 HTTP 超时设置。1000 是一个平衡点单次请求的处理时间在 10 秒左右网络往返开销可以接受。Jev 服务的 workers 数量设为 4因为云主机是 8 核每个 worker 加载一份模型4 个 worker 占用的内存大约 16G留出足够的内存给系统和 MotherDuck 客户端。如果内存更充裕可以增加到 6 个 worker吞吐量还能再提升一些。MotherDuck 侧的查询用了 LIMIT 来限制每次读取的数据量避免一次性拉取太多数据导致内存溢出。同时raw_comments 表的 comment_id 上有主键索引LEFT JOIN 的效率比较高。如果数据量特别大可以考虑在 comment_text 上建全文索引但这次的数据量还没到那个程度。5. 常见问题与排查技巧实录5.1 Jev 服务返回结果格式不一致这是最常见的问题。Jev 模型在处理某些文本时可能会返回非预期的格式。比如对于特别短的文本它可能返回空字符串对于包含特殊符号的文本它可能返回带有多余空格的类别名。如果没有 TypeSafe 约束这些不一致的结果会直接写入数据库导致后续查询和分析出错。我的解决方法是在 Jev 服务的响应处理层加严格的校验。用 Pydantic 的 validator 对 label 做正则匹配只允许 positive、negative、neutral 三个值。如果匹配失败就记录原始返回内容到错误日志并返回一个默认的 neutral 结果。这样虽然会损失一些准确率但保证了整个流程不会中断。后续可以针对这些异常案例做人工复核或模型微调。5.2 MotherDuck 连接超时MotherDuck 的云服务在数据量较大时查询可能会超时。我遇到过一次查询五十万条记录的 comment_text 时连接在 30 秒后断开。后来发现是 MotherDuck 的默认查询超时设置比较短。解决方法是在连接字符串中增加超时参数conn motherduck.connect(md:my_database?timeout120)同时把单次查询的数据量限制在 1000 条以内避免单次查询返回太多数据。如果确实需要处理大量数据就分批查询每批处理完再查下一批。5.3 批量写入时的主键冲突在结果写回阶段如果某条评论被重复分类就会遇到主键冲突。比如第一次分类后结果已经写入但由于某种原因比如脚本中断后重启同一条评论又被分类了一次。如果没有处理冲突写入就会失败。我的解决方法是使用 SQL 的 ON CONFLICT 子句。在插入时指定 comment_id 为主键如果冲突就更新而不是报错。这样即使重复处理最终结果也是最新的不会导致数据重复。5.4 Jev 模型加载失败本地部署 Jev 时模型文件可能因为下载不完整或版本不匹配而加载失败。我遇到过一次Jev 服务启动时报错说模型文件校验失败。后来发现是下载过程中网络中断文件不完整。解决方法是重新下载模型文件并校验文件的哈希值。另外确保 Jev SDK 的版本和模型文件的版本匹配不同版本的模型文件格式可能不同。5.5 性能不达预期如果实测下来速度没有提升可以从几个方面排查。第一检查数据交换通道是否真的减少了网络往返。如果 Jev 服务和 MotherDuck 不在同一区域网络延迟会抵消掉大部分优化效果。第二检查批量大小是否合适。太小会导致频繁的网络请求太大会导致单次处理时间过长。第三检查 Jev 服务的并发配置。如果 workers 数量太少CPU 利用率上不去推理速度就会受限。我整理了一个常见问题速查表方便快速定位问题现象可能原因排查方法解决方案Jev 返回格式不一致模型对特殊文本处理异常查看错误日志中的原始返回加 TypeSafe 校验异常时返回默认值MotherDuck 连接超时查询数据量过大或超时设置过短检查查询返回的行数和耗时分批查询增加超时参数写入时主键冲突重复处理同一条记录检查结果表中是否有重复 comment_id使用 ON CONFLICT 更新Jev 模型加载失败模型文件不完整或版本不匹配校验文件哈希检查 SDK 版本重新下载统一版本速度提升不明显网络延迟高或批量大小不合适检查 Jev 服务和 MotherDuck 的区域同区域部署调整批量大小5.6 实操心得先小规模验证再全量跑这是我最想分享的一条经验。在正式处理五十万条数据之前我先用 100 条数据跑通了整个流程确认每个环节都正常。然后用 1000 条数据测试性能调整批量大小和并发数。最后才用全量数据跑。这样做的好处是如果某个环节有问题在小规模阶段就能发现修复成本很低。如果直接上全量数据一旦中途出错可能需要从头再来浪费大量时间。另外建议在脚本中加入进度日志。每处理完一批数据就打印一次进度这样即使任务运行时间很长也能知道当前进展。如果发现进度停滞可以及时介入排查。6. 改造后的效果与后续扩展方向改造完成后我用同样的五十万条评论数据做了对比测试。原来的流程耗时约三十五分钟改造后的流程耗时约四十二秒。其中数据从 MotherDuck 读取耗时约八秒Jev 推理耗时约二十五秒结果写回耗时约九秒。整体速度提升了约五十倍和标题里说的快五十倍基本吻合。这个提升主要来自两个方面一是消除了数据导出和导入的开销数据不再需要落地到本地文件二是批量处理和并发推理充分利用了计算资源Jev 服务的吞吐量比单条处理时高了很多。后续还可以从几个方向继续优化。第一把 Jev 模型进一步量化减少模型大小和推理时间这样在 CPU 上的推理速度还能再提升。第二把分类结果缓存起来对于重复出现的文本直接返回缓存结果避免重复推理。第三把整个流程封装成 MotherDuck 的定时任务每天自动处理新增的评论数据实现全自动化。我在实际使用中发现TypeSafe 约束虽然增加了一些代码量但它带来的稳定性提升是值得的。尤其是在处理生产数据时数据质量参差不齐没有类型约束的话批量任务很容易因为个别脏数据而失败。另外MotherDuck 的 SQL 接口和 DuckDB 的兼容性很好大部分 DuckDB 的 SQL 语法都可以直接使用这降低了迁移成本。如果你也在做类似的数据密集型推理任务不妨试试这个思路把计算推到数据附近用类型约束保证稳定性用批量处理提升吞吐量。
