AI 数据分析平台架构从脚本烟囱到统一服务层的演进路径大家好我是朱大喜。今天聊一个数据分析团队迟早会撞上的问题当你的 AI 分析需求从跑一次模型变成每天跑一千次模型原来的脚本还能撑多久一、脚本烟囱数据分析的城中村时代不知道你们团队有没有这样的经历——业务方说帮我预测一下下个季度的用户留存率数据分析师吭哧吭哧写个 Python 脚本在本地 Jupyter 里跑完把结果截图发群里。过两天另一个同事要做类似的预测但他不知道有人写过于是又吭哧吭哧写了一版。这就是典型的脚本烟囱模式。每个需求都是一座孤立的违章建筑脚本散落在各自的电脑里逻辑没有复用环境依赖像拆盲盒一样不可控你这个脚本在我这儿报错啊你是不是装了 xxx 库。graph TD A[业务方需求1] -- B[分析师A写脚本] C[业务方需求2] -- D[分析师B写脚本] E[业务方需求3] -- F[分析师C写脚本] B -- G1[本地Jupyter运行] D -- G2[本地Jupyter运行] F -- G3[本地Jupyter运行] G1 -- H1[截图发群] G2 -- H2[截图发群] G3 -- H3[截图发群] style B fill:#f9f,stroke:#333 style D fill:#f9f,stroke:#333 style F fill:#f9f,stroke:#333 style H1 fill:#faa,stroke:#333 style H2 fill:#faa,stroke:#333 style H3 fill:#faa,stroke:#333这种模式的问题就像城中村的水电管道一样——看着能用一旦规模上来就到处漏水重复造轮子同一个数据清洗逻辑被三个同事用三种方式写了三遍还都不太一样可复现性为零上周跑通的脚本这周换了数据源就崩了因为没人知道当初依赖的库版本是什么没有监控和容错凌晨跑的任务挂了直到第二天上班才被发现AI 模型管理失控模型文件散落各处线上跑着的是哪个版本根本没人说得清。二、统一服务层长什么样一个外卖平台的比喻如果把脚本烟囱比作每家都自己做饭吃那统一服务层就是外卖平台。你不用管厨师在哪、用什么锅、几点开工——你只负责下单平台帮你调度、生产、配送。统一服务层的核心组件可以这样划分graph TB subgraph 接入层 A1[Web Dashboard] A2[REST API] A3[定时调度器] end subgraph 服务层 B1[任务编排引擎] B2[计算资源池br/Spark/ClickHouse/Python] B3[AI模型服务br/Model Serving] end subgraph 数据层 C1[数据仓库br/ODS-DWD-DWS-ADS] C2[特征存储br/Feature Store] C3[模型注册中心br/Model Registry] end subgraph 治理层 D1[元数据管理] D2[血缘追踪] D3[监控告警] end A1 -- B1 A2 -- B1 A3 -- B1 B1 -- B2 B1 -- B3 B2 -- C1 B2 -- C2 B3 -- C2 B3 -- C3 B2 -- D1 B3 -- D3 D1 -- D2每一层都有它不可替代的职责接入层负责接收请求不管是手动在 Dashboard 上点运行还是 API 调用还是 crontab 到点了触发服务层是核心大脑决定这个任务用 Spark 算还是用 Python 算、要不要先调模型推理再做聚合数据层让数据和模型不再散落统一存储、统一版本管理治理层盯着一切谁的数据、谁的模型、跑没跑完、报没报错。三、最核心的管道任务编排与调度统一服务层的心脏是任务编排引擎。它的核心工作是回答三个问题这个任务依赖哪些上游上游跑完了吗跑完之后要不要触发下游我们用 Airflow 来搭一个最简单的 DAG展示这个逻辑from airflow import DAG from airflow.operators.python import PythonOperator from datetime import datetime, timedelta default_args { owner: data-team, # 重试次数数据任务难免因为网络抖动失败给 2 次机会 retries: 2, # 重试间隔别马上重试等 5 分钟让上游缓一缓 retry_delay: timedelta(minutes5), # 任务超时防止某个任务卡死把整个调度拖垮 execution_timeout: timedelta(hours2), } dag DAG( daily_user_prediction, default_argsdefault_args, # 每天早上 8 点跑这样上班就能看到结果 schedule_interval0 8 * * *, start_datedatetime(2026, 7, 1), # 不补跑历史数据避免一上线就疯狂回填 catchupFalse, ) def extract_user_features(**context): 从数据仓库拉取用户行为特征 为什么拆成独立节点数据拉取失败的概率最高 独立出来后可以单独重试不用带着模型推理一起重跑 from pyspark.sql import SparkSession spark SparkSession.builder.getOrCreate() # 读 DWS 层的用户宽表这层数据已经按天聚合好了 df spark.sql( SELECT user_id, login_days_30d, avg_session_minutes, payment_times_30d, last_active_days FROM dws.user_behavior_daily WHERE dt ${ds} ) # 转成 Pandas 给 Python 节点用避免序列化问题 return df.toPandas() def run_prediction_model(**context): 调用已注册的 MLflow 模型做推理 为什么用 MLflow 而不是直接 load pkl 文件 MLflow 帮你管理了模型的版本和环境依赖 你不会出现同事模型用 sklearn 1.2 我的是 1.3这种坑 import mlflow import pandas as pd # 从上游任务拿数据XCom 是 Airflow 的任务间通信机制 ti context[ti] features ti.xcom_pull(task_idsextract_features) # 从模型注册中心加载生产力阶段的模型 # 为什么不直接用最新的Production 标签代表经过验证的版本 model mlflow.pyfunc.load_model( model_urimodels:/user_churn_prediction/Production ) predictions model.predict(features) features[churn_prob] predictions return features extract_task PythonOperator( task_idextract_features, python_callableextract_user_features, dagdag, ) predict_task PythonOperator( task_idrun_prediction, python_callablerun_prediction_model, dagdag, ) # 定义依赖关系必须先提取特征才能跑预测 extract_task predict_task注意几个关键的架构决策任务粒度的拆分数据提取和模型推理是两个独立节点。这看起来啰嗦但在生产环境里数据提取失败的概率远高于模型推理拆开后可以只重试失败的节点超时控制每个任务都有execution_timeout防止一个异常任务把整个调度队列卡死模型版本管理通过 MLflow 的Production标签确保线上跑的永远是验证过的版本避免手滑上线了一个实验阶段的模型。四、从脚本烟囱到统一服务层怎么不翻车不要试图一夜之间重构全部。这种整个推倒重来的想法是架构演进最大的敌人。第一步建立数据中间层先别动任何人的脚本。只做一件事把大家都会用到的数据表用户宽表、订单宽表、留存表统一建在数据仓库的 DWS 层写好字段注释和血缘。这一步零风险但价值巨大——至少保证大家用的是同一套数据口径。第二步AI 模型注册中心把所有线上 AI 模型注册到 MLflow 或自建的模型仓库里。每个模型必须有版本号、训练时间、训练数据时间范围、评估指标。目标很简单三个月后你还能找到当时用的是哪个模型。第三步高频任务优先迁移挑团队里跑得最频繁、最痛苦的那 3~5 个分析任务把它们迁移到统一调度平台。选最痛的先动因为痛感是最好的推动力——你能得到最直接的反馈和最强烈的支持。第四步沉淀通用组件当迁移了 10 个任务后你会自然发现有些组件反复出现数据校验、特征工程、结果推送。把它们打包成可复用的 Operator新任务直接调用开发效率会指数级提升。# 沉淀下来的通用数据校验组件 class DataQualityOperator(PythonOperator): 为什么封装成 Operator每个分析任务上线前都应该跑一遍基础校验 与其让每个人自己写不如提供一个统一的钩子 def __init__(self, table_name, checks, **kwargs): checks 示例: { row_count 0: 不能是空表, null_rate(user_id) 0.01: user_id 缺失率不超过 1%, dt last_dt: 分区日期必须是最新的 } super().__init__(task_idfquality_check_{table_name}, **kwargs) self.table_name table_name self.checks checks五、总结从脚本烟囱到统一服务层本质上是一个将个人能力转化为团队能力的过程。脚本烟囱时期效率瓶颈是人——能写代码的人不够用统一服务层阶段效率瓶颈是系统——计算资源和调度能力。这个过程没有银弹核心原则就三条先标准化数据再标准化流程——数据是地基流程是上层建筑让 AI 模型像普通数据一样被管理——版本、血缘、监控缺一不可从最痛的地方开始改——不要为了架构而架构要为了解决实际问题。下一篇文章我们来聊聊 Spark SQL 的 AQE看看怎么让查询引擎自己学会加速。下周见