1. 为什么我要自己搭一个轻量 AI 数据平台去年年底我手上同时压着三个跟模型相关的需求一个要做用户行为日志的清洗和特征回填一个要把业务侧导出的 CSV 做批量推理还有一个要给算法同事提供一个能快速验证新算子的沙箱环境。这三个需求看起来八竿子打不着但真正动手之后我发现它们卡在同一个地方——数据流转的中间层太散了。散到什么程度清洗脚本散在三个仓库里有的用 pandas 写有的用 Spark还有一个是同事用 shell 拼出来的推理任务靠手动传文件中间结果落在本地磁盘跑完一轮要人工核对算子验证更离谱每次都要重新拉一份数据、重新写一遍读取逻辑。这种状态下任何一个新需求进来我都要先花半天时间搞清楚这份数据到底该从哪读、往哪写。所以我决定搭一个轻量的 AI 数据平台初版。注意这里的两个限定词轻量和初版。轻量意味着我不打算一上来就搞微服务、搞 K8s、搞一整套调度系统那玩意儿搭起来三个月起步维护成本还高初版意味着我要的是能跑通核心链路的最小闭环先把数据进来—处理—出去这条路打通后面再逐步加能力。这篇文章我会把整个搭建过程拆开讲架构上我做了哪些取舍、为什么这么选、ETL 和算子这两块具体怎么落地、踩了哪些坑、以及最关键的——这个初版离真正的生产环境到底差在哪。如果你也在做类似的事情或者正准备从零搭一个数据处理中间层这篇应该能帮你少走一些弯路。需要提前说明的是我这里的AI 数据平台不是那种大厂级别的机器学习平台没有特征商店、没有模型注册中心、没有在线推理服务。它更像是一个面向 AI 任务的数据管道 算子执行层核心解决的是数据怎么高效地流到模型面前这个问题。定位清楚了后面的架构决策才不会跑偏。2. 架构决策我为什么放弃了微服务和大数据栈2.1 从能跑倒推架构而不是从先进正推刚开始规划的时候我列了一堆候选方案Kafka Flink Hive 的经典大数据栈、基于 FastAPI 的微服务拆分、甚至考虑过直接上 Airflow 做调度。但冷静下来问自己一个问题我现在的数据量级是多少日增日志大概 20GB单次批量推理的数据量在百万行级别算子验证的数据量更小。这个量级用大数据栈纯属杀鸡用牛刀。我最终选的是单体 模块化的架构。具体来说一个 Python 服务进程内部按职责分成四个模块接入层、ETL 层、算子执行层、存储层。模块之间通过明确的接口通信但不拆成独立服务。这样做的好处很直接部署简单一个进程搞定、调试方便不用跨服务追日志、资源占用低一台 16 核 64G 的机器足够跑。提示轻量平台最大的敌人不是性能而是过度设计。我见过太多项目在初期就引入消息队列和分布式调度结果数据量根本没到那个级别反而被运维复杂度拖垮。2.2 存储选型的取舍为什么是 Parquet DuckDB存储这块我纠结了很久。候选有四个MySQL、ClickHouse、Parquet 文件 DuckDB、以及直接上对象存储。MySQL 的问题是列式分析性能差做聚合查询时全表扫描很慢ClickHouse 性能好但运维成本高而且我的场景里写入和查询都不算高频用不上它的实时能力对象存储适合冷数据但做中间结果交换时延迟太高。最后我选了Parquet 文件 DuckDB的组合。Parquet 负责列式存储压缩比高、读取快而且和 pandas、Polars 这些库的兼容性极好DuckDB 负责查询它是一个嵌入式的分析型数据库不需要独立部署直接读 Parquet 文件就能跑 SQL性能在百万行级别完全够用。这个组合最大的好处是零运维。没有数据库进程要管没有连接池要调数据就是文件查询就是进程内的计算。对于初版来说这是性价比最高的选择。2.3 算子执行层为什么我坚持要抽象出算子这个概念这是整个平台里我最坚持的一个设计。所谓算子就是一个可复用的数据处理单元输入是一张表或一个数据流输出是处理后的结果。比如去重字段映射窗口聚合特征归一化每一个都可以是一个独立算子。为什么要抽象这一层因为 AI 任务的数据处理有大量重复逻辑。今天做用户画像要算活跃度明天做推荐要算点击率后天做风控要算异常分数这些计算背后的数据处理模式高度相似。如果每次都从头写脚本代码复用率极低而且容易出错。抽象成算子之后带来三个直接收益第一算子可以独立测试输入输出明确测试用例好写第二算子可以组合一个 ETL 流程就是若干个算子的串联第三算子可以独立优化比如某个聚合算子性能不行我可以单独把它换成向量化实现不影响其他部分。2.4 调度我用了最土但最稳的方案调度这块我没上 Airflow也没上 Celery用的是基于文件锁的简单任务队列 cron 触发。具体逻辑是任务定义写在一个 YAML 文件里cron 每分钟扫一次发现有到期的任务就加锁执行执行完写状态文件。听起来很土对吧但它有几个实实在在的好处没有额外的调度服务要维护、任务状态一目了然就是文件、出问题容易排查看日志和状态文件就行。对于初版来说稳定压倒一切。等任务量和复杂度上来了再迁移到专业调度系统也不迟。3. ETL 链路从原始数据到可用特征3.1 数据接入统一入口比什么都重要ETL 的第一步是数据接入。我这里的原始数据来源有三种业务库导出的 CSV、日志文件JSON Lines 格式、以及上游系统推送的 Parquet 文件。如果每种来源都写一套读取逻辑代码会迅速膨胀。我的做法是定义一个统一的接入接口所有数据源都要实现read()方法返回一个标准的 DataFrame。CSV 用 pandas 读JSON Lines 用逐行解析后转 DataFrameParquet 直接用 DuckDB 读。读取之后统一做三件事字段名标准化统一转小写下划线、类型推断和转换、以及元数据登记记录数据来源、行数、字段列表。这里有个细节值得说类型推断一定要显式做不能依赖自动推断。我踩过一个坑某次 CSV 里的 ID 字段被自动推断成了整数结果前导零全丢了下游关联时对不上。后来我强制要求每个数据源配置里声明字段类型宁可多写几行配置也不要让自动推断埋雷。3.2 清洗算子去重、空值、异常值三板斧数据接入之后就是清洗。我把清洗拆成了三个基础算子每个都可以独立配置和复用。去重算子支持两种模式全字段去重和指定主键去重。全字段去重直接用drop_duplicates()主键去重则保留最新一条按时间戳排序。这里要注意去重前一定要确认主键的唯一性语义我遇到过主键重复但业务上合法的场景盲目去重会丢数据。空值处理算子提供四种策略删除、填充默认值、前向填充、以及标记为特殊值。选择哪种策略取决于字段语义。比如用户年龄缺失填充中位数比删除更合理但如果是关键的业务 ID 缺失那只能删除。异常值检测算子用的是 IQR四分位距方法对数值型字段计算上下界超出范围的标记为异常。这里我加了一个配置项允许对每个字段单独设置是否启用异常检测因为有些字段比如金额的极端值本身就是有意义的信号不能一刀切。3.3 特征工程算子把重复劳动变成配置特征工程是 ETL 里最耗时的部分。我抽象了几个高频算子窗口聚合算子按指定分组字段和时间窗口做聚合支持 sum、mean、count、max、min 等聚合函数。实现上用 pandas 的groupbyrolling对于大数据量场景可以切换到 Polars。特征交叉算子把两个或多个字段组合成新特征支持拼接、哈希、以及分桶交叉。归一化算子支持 min-max、z-score、以及 robust 三种归一化方式。时间特征算子从时间戳提取年、月、日、小时、星期等特征还支持计算时间差。这些算子组合起来一个典型的用户特征表可以在半小时内配置完成而之前手写脚本至少要一天。3.4 ETL 流程编排YAML 配置驱动的流水线单个算子写好之后怎么把它们串起来我用的是 YAML 配置。一个 ETL 流程定义大概长这样name: user_feature_pipeline source: type: csv path: /data/raw/user_events.csv schema: user_id: string event_time: timestamp event_type: string amount: float steps: - operator: dedup params: keys: [user_id, event_time] - operator: fillna params: amount: 0.0 - operator: window_agg params: group_by: user_id window: 7d aggs: amount: [sum, mean, count] - operator: normalize params: columns: [amount_sum] method: minmax sink: type: parquet path: /data/features/user_features.parquet这个配置驱动的方式有几个好处流程可读一眼看出数据怎么流转、可版本管理YAML 进 Git、可复用改几个参数就是新流程。当然也有代价就是灵活性不如直接写代码遇到特别复杂的逻辑还是得写自定义算子。4. 算子执行层性能、并发和资源控制4.1 算子执行的三种模式算子执行层我设计了三种模式对应不同的数据规模和延迟要求。单机内存模式数据量小于 1GB 时直接用 pandas 在内存里算简单直接延迟最低。这是默认模式。分块流式模式数据量在 1GB 到 50GB 之间时把数据按行分块逐块读取、逐块计算、逐块写出。这样内存占用可控不会 OOM。实现上用 pandas 的chunksize参数或者 Polars 的 lazy 模式。DuckDB 外存模式数据量超过 50GB 或者需要复杂 SQL 时交给 DuckDB 处理。DuckDB 会自动做外存溢写支持比内存大的数据集而且 SQL 表达能力强复杂聚合写起来比 pandas 清爽。三种模式的切换是自动的根据数据源的大小和算子类型判断。这个自动切换逻辑我调了很久核心是宁可保守也不要激进因为一旦 OOM整个任务就挂了而多花点时间用流式模式至少能跑完。4.2 并发控制为什么我用了进程池而不是线程池Python 的 GIL 是个绕不开的问题。算子执行是 CPU 密集型任务用线程池根本跑不满多核。所以我用的是进程池每个算子任务分配一个独立进程通过共享内存或临时文件交换数据。进程池的坑在于数据序列化开销。如果每个任务都要把 DataFrame 序列化传过去开销会很大。我的优化是对于大 DataFrame先落盘成 Parquet进程间只传文件路径对于小 DataFrame直接用 pickle 传。这个阈值我设的是 100MB超过就落盘。并发度也不是越高越好。我实测下来并发度设成 CPU 核数的 1.5 倍左右比较合适再高会因为上下文切换和内存竞争导致性能下降。这个值可以在配置里调不同机器不一样。4.3 算子性能优化向量化是王道写算子的时候最大的性能陷阱是逐行循环。我见过太多用iterrows()或者apply()逐行处理的代码在百万行数据上慢得令人发指。正确的做法是向量化。能用 pandas 内置函数的绝不用循环能用 numpy 数组运算的绝不用 pandas 的 apply。举个例子计算两个字段的比值df[a] / df[b]比df.apply(lambda r: r[a] / r[b], axis1)快几十倍。对于确实需要复杂逻辑的场景我推荐用numpy 的向量化操作或者numba 的 JIT 编译。numba 可以把 Python 函数编译成机器码对于数值计算密集的算子加速比能达到 10 倍以上。代价是首次编译有开销而且不是所有 Python 语法都支持。4.4 资源隔离别让一个任务拖垮整个平台初版虽然简单但资源隔离还是要做的。我的做法是给每个任务设置内存上限和超时时间。内存上限通过resource模块的setrlimit实现超时通过信号机制实现。一旦任务超限直接杀掉记录日志不影响其他任务。这个机制救过我好几次。有一次一个算子写错了产生了笛卡尔积内存瞬间飙升如果没有限制整台机器都会被拖垮。加上限制之后任务被杀掉我收到告警改完代码重新跑就行。5. 踩坑实录那些让我熬夜的瞬间5.1 时间戳时区问题一个让数据对不上的隐形杀手这是我最想吐槽的一个坑。原始数据里的时间戳有的是 UTC有的是本地时间而且没有明确标记。我一开始没注意直接按字符串解析结果做窗口聚合时同一批数据被分到了不同的时间窗口特征全乱了。排查过程很痛苦因为数据本身没有报错只是结果不对。我最后是通过对比两个数据源的原始时间戳和聚合结果才发现时区不一致。解决方案是所有时间戳在接入时统一转成 UTC并在元数据里记录原始时区。转换用pandas.to_datetime()配合tz_localize和tz_convert虽然多了一步但避免了后续所有的时间相关 bug。注意时间戳处理是数据平台里最容易出错的地方之一。如果你的数据源跨时区一定要在接入层就统一不要留到下游处理。5.2 内存泄漏进程池用久了会吃光内存进程池有个隐蔽的问题如果子进程里创建了全局对象或者缓存这些对象在进程复用时会一直累积导致内存泄漏。我遇到的情况是某个算子里用了一个全局的缓存字典跑了几百个任务之后内存从 2GB 涨到了 30GB。排查方法是给进程池加上maxtasksperchild参数让子进程执行一定数量的任务后就重启。这个参数设成 50 左右比较合适既能复用进程减少启动开销又能定期清理内存。另外算子代码里要避免使用全局可变对象所有状态都通过参数传递。5.3 文件句柄耗尽小文件太多导致的连锁反应Parquet 文件如果分片太多每个分片都要打开一个文件句柄任务一多就会耗尽系统的文件句柄限制。我遇到过一次任务跑到一半报Too many open files整个流程中断。解决方案有两个层面第一合并小文件在写入时控制分片大小避免产生大量小文件第二提高文件句柄限制通过ulimit -n调整同时在代码里用上下文管理器确保文件及时关闭。我现在的做法是写入时按 128MB 一个分片读取时用 DuckDB 的通配符读取它会自动处理多文件。5.4 算子版本管理改了算子历史任务结果对不上这个问题比较隐蔽。我优化了一个归一化算子改了默认的归一化方式结果发现之前跑的历史任务结果和现在跑的对不上。因为任务配置里没有记录算子版本无法复现历史结果。后来我加了一个机制每个算子在注册时带上版本号任务配置里记录使用的算子版本。算子升级时保留旧版本新任务用新版本历史任务可以按记录的版本复现。这个机制在调试和审计时非常有用。6. 初版和生产环境的真实差距6.1 可靠性从能跑到跑不挂初版最大的问题是没有容错机制。任务失败了就是失败了需要人工重跑。生产环境需要的是自动重试、断点续跑、以及失败告警。具体来说生产级平台需要任务级别的重试策略比如失败重试 3 次每次间隔递增、算子级别的断点续跑记录每个算子的执行状态失败后从断点继续、以及完善的监控告警任务失败、超时、资源超限都要有告警。我现在的初版只做了最基础的重试断点续跑还没实现。这是下一步要补的重点。6.2 可观测性日志、指标、链路追踪一个都不能少初版的日志是散落在各个算子里的 print 和 logging排查问题时要在大量日志里翻找。生产环境需要的是结构化日志 指标采集 链路追踪。结构化日志意味着日志是 JSON 格式包含任务 ID、算子名称、执行时间、输入输出行数等字段可以直接被日志系统采集和分析。指标采集包括任务成功率、平均执行时间、资源使用率等用于监控和容量规划。链路追踪则是记录一个任务从开始到结束的完整调用链方便定位性能瓶颈。6.3 数据质量从事后发现到事前拦截初版对数据质量基本没有校验都是跑完发现结果不对再回头查。生产环境需要在数据流转的每个环节做质量检查接入时检查字段完整性和类型正确性算子执行后检查输出行数和数值范围写出前检查数据一致性。我现在的做法是在关键节点加了一些断言但还比较零散。完整的方案应该是定义一个数据质量规则集每个数据集关联一组规则任务执行时自动校验不通过就阻断并告警。6.4 扩展性单机到分布式的鸿沟初版是单机的所有算子都在一台机器上跑。数据量再涨一个数量级单机就扛不住了。从单机到分布式需要解决数据分片、任务调度、状态同步、以及网络传输等一系列问题。我的规划是分两步走第一步先把算子执行层抽象成可远程调用的服务数据通过共享存储交换第二步再引入分布式调度框架把任务分发到多台机器。这个过程不能急每步都要保证稳定性。6.5 安全与权限初版完全没有考虑的部分说实话初版在安全方面是空白的。没有用户认证没有权限控制没有数据脱敏。这在内部小范围使用没问题但一旦开放给更多人用就是隐患。生产环境需要用户认证和授权谁能提交任务、谁能访问哪些数据、数据脱敏敏感字段在非生产环境要脱敏、以及操作审计谁在什么时候做了什么操作。这些在初版里都没做是后续必须补的。7. 如果重来一次我会怎么调整7.1 先定义数据契约再写代码初版我是边写代码边想数据格式导致后期改字段名、改类型时到处改代码。如果重来我会先定义好数据契约每个数据集的字段名、类型、语义、以及质量约束写成 schema 文件代码根据 schema 自动生成校验和转换逻辑。这样数据格式变更时只需要改 schema代码不用动。7.2 算子从第一天就带版本和测试算子版本管理我是后期才加的导致前期的一些算子没有版本记录历史结果无法复现。如果重来我会从第一个算子开始就带版本号并且每个算子都配单元测试。测试用例覆盖正常输入、边界输入、以及异常输入确保算子行为稳定。7.3 监控和告警前置而不是后置初版我是等出了问题才加监控属于被动响应。如果重来我会在搭平台的同时就把监控和告警搭起来至少包括任务成功率、执行时间、资源使用率这几个核心指标。这样问题能在第一时间被发现而不是等用户反馈。7.4 文档和示例同步写别攒着初版的功能文档我是最后补的结果很多细节自己都记不清了只能翻代码。如果重来我会每完成一个功能就写文档和示例保持文档和代码同步。这不仅方便别人用也方便自己回顾。8. 一些实操层面的小技巧8.1 用 DuckDB 做数据探查比 pandas 快得多调试 ETL 流程时经常需要快速看一下数据长什么样。用 pandas 读大文件很慢用 DuckDB 直接SELECT * FROM file.parquet LIMIT 10秒出结果。我现在的习惯是任何数据探查先用 DuckDB 跑一遍确认数据没问题再写正式代码。8.2 算子开发时先用小数据集验证逻辑写算子时不要一上来就用全量数据跑先用几百行的小数据集验证逻辑正确性。小数据集跑得快改起来也快。逻辑验证通过后再用全量数据测性能。这个习惯帮我省了大量等待时间。8.3 用 pytest 做算子测试fixture 管理测试数据算子的单元测试我用 pytest测试数据用 fixture 管理。每个算子配一个 fixture提供标准输入和预期输出。这样测试代码简洁而且测试数据可以复用。对于复杂的算子我还会加参数化测试覆盖不同的配置组合。8.4 日志里带上任务 ID 和算子名称排查时能省一半时间初版的日志没有统一格式排查问题时要在大量日志里翻找。后来我强制要求所有日志都带上任务 ID 和算子名称格式统一成[task_id][operator] message。这样用 grep 一搜就能定位到具体任务的日志排查效率提升明显。8.5 配置文件用 YAML 而不是 JSON可读性好太多JSON 不支持注释写配置时想加个说明都不行。YAML 支持注释、支持多行字符串、可读性也好。我现在的所有配置都是 YAML包括任务定义、算子参数、数据源配置。唯一的坑是 YAML 的缩进敏感写的时候要注意对齐。8.6 定期清理中间结果别让磁盘爆掉ETL 流程会产生大量中间结果如果不清理磁盘很快就满了。我的做法是给中间结果设置 TTL比如 7 天用一个定时任务定期清理过期的中间文件。同时重要的中间结果会标记为持久化不参与清理。这个机制简单但有效避免了磁盘告警。搭这个平台的过程让我最大的体会是轻量不等于简陋初版不等于凑合。每一个架构决策背后都要有明确的理由每一个坑都要记录下来避免重蹈覆辙。这个初版现在还在跑每天处理几十个任务虽然离生产级还有距离但它已经实实在在解决了我的问题。后面我会继续迭代把可靠性、可观测性、数据质量这几块补上让它真正能扛住生产环境的考验。
