AI 数据分析平台架构：从脚本烟囱到统一服务层的演进路径
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(minutes=5), # 任务超时：防止某个任务卡死把整个调度拖垮 'execution_timeout': timedelta(hours=2), } dag = DAG( 'daily_user_prediction', default_args=default_args, # 每天早上 8 点跑，这样上班就能看到结果 schedule_interval='0 8 * * *', start_date=datetime(2026, 7, 1), # 不补跑历史数据，避免一上线就疯狂回填 catchup=False, ) 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_ids='extract_features') # 从模型注册中心加载"生产力"阶段的模型 # 为什么不直接用最新的：Production 标签代表经过验证的版本 model = mlflow.pyfunc.load_model( model_uri='models:/user_churn_prediction/Production' ) predictions = model.predict(features) features['churn_prob'] = predictions return features extract_task = PythonOperator( task_id='extract_features', python_callable=extract_user_features, dag=dag, ) predict_task = PythonOperator( task_id='run_prediction', python_callable=run_prediction_model, dag=dag, ) # 定义依赖关系：必须先提取特征，才能跑预测 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_id=f'quality_check_{table_name}', **kwargs) self.table_name = table_name self.checks = checks五、总结
从脚本烟囱到统一服务层，本质上是一个将个人能力转化为团队能力的过程。脚本烟囱时期，效率瓶颈是人——能写代码的人不够用；统一服务层阶段，效率瓶颈是系统——计算资源和调度能力。
这个过程没有银弹，核心原则就三条：
- 先标准化数据，再标准化流程——数据是地基，流程是上层建筑；
- 让 AI 模型像普通数据一样被管理——版本、血缘、监控缺一不可；
- 从最痛的地方开始改——不要为了架构而架构，要为了解决实际问题。
下一篇文章我们来聊聊 Spark SQL 的 AQE，看看怎么让查询引擎"自己学会加速"。下周见！