聊《一个大数据项目改成 AI 流程后，最难的部分完全变了》之前，先说一句实在的：别急着背概念，先看它在真实项目里到底解决什么问题。
摘要
前两年我还在写 MapReduce 和 Spark SQL，觉得数据清洗、数仓建模是硬通货。今年开始带团队做 RAG（检索增强生成）落地，发现最头疼的不是怎么调参让模型更聪明，而是怎么保证它在生产环境里不乱说话、不泄露数据、出事了能查清是谁干的。
很多从大数据转 AI 的朋友，简历上堆满了 LangChain、向量数据库，面试也能聊得头头是道。但真到了上线那一刻，Demo 丝滑，一上生产就崩。为什么？因为传统数据工程关注的是“准确性”和“吞吐量”，而大模型工程在现阶段的核心痛点是“可控性”和“可观测性”。
如果你只盯着 Prompt 调优或者检索率，忽略了下层的权限隔离和日志审计，那你的 AI 项目永远只是个玩具。今天复盘一个我们刚上线的项目，聊聊数据工程师在这个转型期，到底该修哪些“新基本功”。
目录
- 交叉点：从 ETL 到 ETR，思维模式的根本转变
- 数据治理的新维度：元数据与权限标签
- 向量数据库：不仅仅是存储，更是过滤网
- RAG 数据管道：日志与可观测性的实战
- 落地项目建议：先做减法，再做加法
- 总结
交叉点：从 ETL 到 ETR，思维模式的根本转变
在传统大数据领域，我们的核心工作流是 ETL（Extract, Transform, Load）。输入是结构化的日志或业务数据，输出是准确的报表或特征表。这个过程讲究确定性：同样的输入，必须得到同样的输出。
但在大模型时代，尤其是涉及 Agent 或 RAG 架构时，流程变成了 ETR（Extract, Transform, Retrieve/React）。这里引入了两个巨大的变量：
1. 非确定性：模型的输出具有概率性，即使 Prompt 完全一样，每次结果可能微调。
2. 黑盒交互：模型会通过 API 调用外部工具（如查询数据库、修改用户信息），这种“行动”一旦失控，后果比报表错误严重得多。
我见过很多数据工程师，把向量数据库当成普通的 NoSQL 数据库来用，只管往里面塞数据，不管数据的权限归属。结果就是，普通员工通过 RAG 应用，竟然能检索出 CEO 的薪资数据，或者通过 Agent 执行了删除订单的操作。
所以，转型的第一步不是学 Python 框架，而是建立“安全边界意识”。在数据管道中，不仅要清洗数据，还要清洗“权限上下文”。
数据治理的新维度：元数据与权限标签
在传统数仓中，我们治理的是数据质量（完整性、一致性）。在 AI 场景下，我们必须给每一段知识数据打上“权限标签”和“可信度标签”。
比如，我们的内部文档库里有 10 万份文档。以前我们只关心分词效果好不好，现在我们要关心：这份文档属于哪个部门？谁有权访问？它是否包含敏感PII（个人身份信息）？
我们在项目中引入了一套简单的元数据治理方案。在向量化之前，增加一个中间层，对原始数据进行打标：
class DataProcessor: def __init__(self, db_client): self.db = db_client # 加载权限映射表，假设存储在关系型数据库中 self.perm_map = self.load_permission_matrix() def process_and_tag(self, raw_text: str, source_doc_id: str) -> dict: """ 处理原始文本并附加权限和可信度标签 """ # 1. 基础清洗 clean_text = self.clean(raw_text) # 2. 获取来源权限 doc_meta = self.db.get_metadata(source_doc_id) allowed_roles = doc_meta.get('allowed_roles', []) # 3. 简单敏感词过滤（生产环境应使用专门的 PII 识别模型） is_sensitive = self.check_pii(clean_text) return { "content": clean_text, "metadata": { "source_id": source_doc_id, "roles": allowed_roles, "is_sensitive": is_sensitive, "vector_ready": True } }这段代码看起来简单，但它解决了两个大问题：
1. 数据隔离：向量数据库本身很难做细粒度的行级权限控制。我们将权限信息嵌入 metadata，在检索阶段进行二次过滤。
2. 质量追溯：is_sensitive标记可以帮助我们在后续分析中，剔除那些因隐私合规问题导致的数据噪声。
对于数据工程师来说，这就是你的新“数据血缘”。你不仅要追踪数据是从哪张表来的，还要追踪这份数据片段被哪些角色有权消费。
向量数据库：不仅仅是存储，更是过滤网
很多人误以为向量数据库（如 Milvus, Pinecone, pgvector）只是个存向量的仓库。其实，在生产环境中，它更像是一个带条件的过滤器。
传统的 SQL 查询有WHERE user_id = ?，向量查询通常只有ANN(query_vector)。但这不够。我们需要的是ANN(query_vector) AND role IN ('admin', 'user')。
我的建议是：永远不要信任向量数据库的内置权限系统。大多数向量库的权限管理非常粗糙，或者根本不具备企业级的 RBAC（基于角色的访问控制）。
最佳实践是“两级过滤”：
1. 预过滤（Pre-filtering）：在发起向量检索前，先在关系型数据库或缓存中查出该用户有权访问的文档 ID 集合（或使用布隆过滤器判断交集）。将这部分信息作为filter参数传给向量库。
2. 后过滤（Post-filtering）：如果向量库返回的结果中混入了无权数据，必须在应用层截断，绝不回显给用户。
这里有个坑：如果你的权限集合很大（比如全公司 5000 人），传给向量库的过滤条件会极其复杂，导致检索性能骤降。这时候，你需要利用数据工程的老本行——分区。按部门或项目组对向量索引进行物理分区，检索时只查对应的分区。
RAG 数据管道：日志与可观测性的实战
这是本文最想强调的部分。Demo 跑通很容易，但上线后，你怎么知道模型为什么胡说八道？怎么知道 Agent 是否被恶意诱导？
传统大数据监控看的是 Job 是否失败、延迟多少毫秒。AI 应用的监控要看的是：Trace（调用链）。
我们需要记录每一次用户提问的完整生命周期：
1. User Input：用户问了什么？
2. Context Retrieval：检索到了哪些文档？它们的得分是多少？
3. Model Response：模型生成了什么？
4. Tool Call：Agent 调用了什么 API？参数是什么？
如果没有这些日志，当业务方投诉“模型给出了错误代码”时，你只能干瞪眼。你甚至不知道是因为检索没搜到相关文档，还是因为 Prompt 引导错了方向。
我们构建了一个基于 OpenTelemetry 的简易日志框架，关键代码如下：
import logging import uuid logger = logging.getLogger("rag_engine") def retrieve_and_generate(user_query, user_id): trace_id = str(uuid.uuid4()) logger.info(f"[TRACE_ID:{trace_id}] Start query from user:{user_id}") try: # 1. 检索 docs = vector_db.search(user_query, top_k=5) logger.info(f"[TRACE_ID:{trace_id}] Retrieved {len(docs)} docs") # 2. 构造 Prompt prompt = build_prompt(user_query, docs) # 3. 调用模型 response = llm.chat(prompt) # 4. 记录关键指标 logger.info(f"[TRACE_ID:{trace_id}] Success. Tokens used: {response.usage}, Content Preview: {response.text[:50]}...") return response except Exception as e: # 异常必须详细记录，包括上下文 logger.error(f"[TRACE_ID:{trace_id}] Failed. Error: {str(e)}, Query: {user_query}") raise这些日志不能只存在服务器本地。你需要将它们推送到 ELK 或 Grafana Loki 中。更重要的是，你要建立一套评分机制：人工标注哪些回答是好是坏，将这些标注数据回流到训练集或 Prompt 优化库中。这才是数据工程师价值最大的地方——数据闭环。
落地项目建议：先做减法，再做加法
如果你正准备转型，或者在公司内部推动 AI 项目，我有几条基于踩坑经验的具体建议：
1. 从“只读”开始：不要一上来就做能修改数据库的 Agent。先做纯检索问答，验证权限控制和日志记录的可行性。
2. 重视“坏数据”：在大数据领域，坏数据影响报表准确性；在 AI 领域，坏数据（幻觉）可能导致法律风险。建立专门的人工审核环节，哪怕只审核 1% 的流量，也能快速发现系统性偏差。
3. 不要过度依赖开源框架的黑盒：LangChain 等框架提供了便利，但也隐藏了复杂性。作为数据工程师，你应该理解底层是如何组装 Token 的，是如何处理长上下文的。遇到 Bug 时，框架的 Issue 列表往往解决不了你的特定业务场景。
4. 简历亮点转变：别再只写“搭建了 Hadoop 集群”或“优化了 SQL 查询”。要写“设计了基于 RBAC 的 RAG 权限隔离方案，实现了 100% 的数据访问审计”、“构建了向量检索的可观测性链路，将排查模型幻觉的时间从小时级降低到分钟级”。
总结
从大数据到大模型，技术栈确实在变，但工程师的核心素养没有变：对数据的敬畏，对流程的控制，对异常的处理。
之前的“脏活累活”是写脚本清洗 CSV，现在的“脏活累活”是设计权限模型、配置日志采集、清洗有毒的 Prompt 数据。这些工作枯燥、不起眼，甚至不如调参那么光鲜，但它们决定了你的 AI 系统是能稳定服务百万用户，还是在上线第一天就被停服整改。
别急着去学 Transformer 的数学原理，先把你的数据管道变成“带护栏的高速公路”。这，才是数据工程师进入 AI 时代的真正门票。
资料展示
下面是我整理的AI大模型学习资料和工具包预览，适合收藏后按主题逐步学习。
如果你想看完整资料目录，可以在评论区留言「资料」；也欢迎告诉我你更关注AI大模型里的哪类内容。
)
