简介一套基于Python后端与Vue3前端构建的ezdata数据处理分析与任务调度系统资源包面向需要统一管理多数据源、构建数据模型并集成LLM智能问答与低代码能力的平台开发及数据工程人员。包内完整呈现前端工程与后端服务覆盖分布式Pandas引擎、TB级数据处理、DAG工作流调度等核心模块可用于系统学习架构设计或直接作为二次开发基座。资源共2000个文件约9.32MB以759个vue、629个ts、263个py文件为主体Vue负责界面组件、TS承载交互逻辑、Python实现数据接入与调度服务前端页面与后端逻辑清晰分离另含png/svg图标及less样式、yaml部署配置等便于按目录模块检索。已有38人学习下载适合中高级开发者快速了解ezdata的整体实现思路与代码组织方式从而缩短同类数据中台项目的搭建周期。对需要梳理数据中台核心链路的读者可结合前端交互与后端服务对照研读降低理解成本。1. ezdata 是什么从多源接入到自动调度的数据平台很多数据团队的日常是数据散落在 MySQL、ClickHouse、Excel 和一堆 CSV 里每天靠人肉跑脚本汇成一张大表再用 cron 把结果推给业务。ezdata 这类系统想解决的就是把这个过程变成配置化流程。它不是一个 BI 报表工具而是一个「多数据源接入 统一数据模型 分布式 Pandas 计算 LLM 智能问答 DAG 工作流调度」的一体化平台Python 后端负责连接数据源、执行分析和编排任务Vue3 前端把数据目录、任务流、问答界面呈现给使用者。如果你团队里有几十张表、几百个 Python 脚本要维护这个方向就是把这堆脚本资产化、可视化、可调度化。适合正在搭内部数据平台被多数据源和定时任务搞到头大的数据工程团队。2. 多数据源管理与统一数据模型先想清楚再做适配器数据源接入是地基。地基设计错了后面所有分析、LLM 问答和调度都会跟着翻车。这一章先讲接入层的适配器怎么设计再讲统一数据模型和元数据持久化——这三件事是一体的分开做最后一定对不上。2.1 多数据源接入层适配器模式与连接管理常见做法是定义一个 DataSourceAdapter 基类MySQL、PostgreSQL、ClickHouse、CSV 文件都实现同一组接口。接口不用太复杂读、写、连接测试、关闭四个能力就够。别贪多接口每多一个方法新增数据源的成本就高一截。第一read 不用流式的大查询做默认入口而是带 limit 的批量抓取。你以为用户只查一万行结果一条 SQL 把 5000 万行拉回来了。limit 参数能在源头拦住这种查询不会让后端进程瞬间涨到几个 G 内存。第二连接不归任务管归连接池管。每个任务自己新建连接是最容易踩的坑并发一高数据库连接数直接被打满。我一般用 DBUtils 的 PooledDB把 maxconnections、maxcached、maxoverflow 三个参数暴露到数据源配置里。下面是一个最小实现from abc import ABC, abstractmethod from typing import Any, Iterable, Iterator from dbutils.pooled_db import PooledDB class DataSourceAdapter(ABC): 所有数据源适配器的基类。新增数据源时只需要实现这四个方法。 def __init__(self, config: dict): self.config config self._pool: PooledDB | None None abstractmethod def connect(self) - Any: 从连接池获取一个连接。 abstractmethod def read(self, query: str, limit: int 10000) - list[dict]: 按查询语句读取数据返回最多 limit 行。 abstractmethod def write(self, rows: Iterable[dict], table: str, batch_size: int 500) - int: 批量写入返回写入行数。 abstractmethod def test_connection(self) - bool: 连通性探活任务开始前调用。代码逻辑说明connect 从连接池拿连接而不是每次新建read 返回 list[dict]让调用方直接拿数据做 pandas 转换write 按 batch_size 分批提交避免一次写入事务过大。limit 参数是安全阀默认 10000不是业务上「应该」只要这么多而是防止手滑全表扫描。参数说明PooledDB 里 maxconnections 是池上限maxcached 是空闲缓存数maxoverflow 是超过 maxconnections 后还能再开的连接数blockingTrue 表示池满时请求排队。这三个值的经验组合是 8/4/2数据源多的系统把 maxconnections 压到 46宁可让任务排队也不要打爆数据库。以 MySQL 为例实现类长这样import pymysql class MysqlAdapter(DataSourceAdapter): def connect(self): if self._pool is None: self._pool PooledDB( creatorpymysql, maxconnections8, maxcached4, maxoverflow2, blockingTrue, **self.config[connect_args], ) return self._pool.connection() def read(self, query, limit10000): conn self.connect() try: with conn.cursor() as cur: cur.execute(query) rows cur.fetchmany(sizemin(limit, 1000)) result [] while rows and len(result) limit: column_names [d[0] for d in cur.description] for row in rows: result.append(dict(zip(column_names, row))) rows cur.fetchmany(size1000) return result[:limit] finally: conn.close()代码说明read 里用 fetchmany(size1000) 每次只取 1000 行内存占用和网络往返都有控制。conn.close() 在连接池场景下是归还连接不是真断开所以 finally 里一定要写。很多人漏了这一步任务跑完连接不还几百个任务之后池子就空了。这里补充一个进阶选择如果你确实要流式读取大结果集可以把 read 改成生成器函数但生成器在调用方提前 break 时finally 的归还时机不可控很容易把连接池搞出泄漏。我的建议是默认就用返回列表的版本流式需求单独开一个 read_stream 方法并明确要求调用方用 with 块管理生命周期。2.2 统一数据模型字段映射与类型收敛规则接入做完了第二步是统一数据模型。为什么要统一源库里 MySQL 的 datetime、ClickHouse 的 DateTime64、PostgreSQL 的 timestamp 明明是一个东西到 pandas 里却要写三套处理逻辑。统一模型就是建一张「内部类型契约表」让下游只认一套类型。常见做法是把 ezdata 内部类型收敛成五个string、int、float、datetime、bool。每个数据源适配器负责把自己的类型映射到这五类TYPE_MAP { mysql: { datetime: datetime, varchar: string, bigint: int, decimal: float, tinyint: bool, text: string, }, clickhouse: { DateTime64: datetime, String: string, Float64: float, Int64: int, Decimal128: float, }, } def normalize_field_type(source: str, source_type: str) - str: mapped TYPE_MAP.get(source.lower(), {}).get(source_type.lower()) return mapped or string逻辑说明normalize_field_type 的返回值就是 ezdata 内部类型下游的 pandas 读取、前端表格渲染、LLM 问答全部只看这个类型。未知类型统一降级成 string这是类型收敛规则里的铁律宁可宽不可猜。猜错类型会导致 pandas 读取时抛异常整个任务失败降级成 string 最多是后续再做一次转换不会崩。这里有一个容易被忽略的细节decimal 直接映射成 float 会有精度损失。涉及金额的字段建议在元数据里标记为 amount映射成 string计算时用 Decimal。否则你算出来的销售额总和后面挂着 0.00000004 的尾巴业务方一眼就看出不对。数据模型另外一层是三层结构datasource数据源→ dataset数据集/表→ field字段。dataset 记录表名、主键、分区键、采样行数field 记录字段名、内部类型、注释、是否参与 LLM 问答。这套模型不仅是给 pandas 用的也是后面 LLM 问答的“食材”。2.3 元数据持久化与数据字典元数据持久化的作用有两个一是让前端能展示数据目录二是给 LLM 提供字段上下文。核心表就三张其中 field 表最关键CREATE TABLE dataset_field ( id BIGINT PRIMARY KEY AUTO_INCREMENT, dataset_id BIGINT NOT NULL, field_name VARCHAR(255) NOT NULL, field_type VARCHAR(32) NOT NULL, field_comment VARCHAR(1024), is_primary_key BOOLEAN DEFAULT FALSE, enable_llm BOOLEAN DEFAULT TRUE, UNIQUE KEY uk_dataset_field (dataset_id, field_name) ) ENGINEInnoDB DEFAULT CHARSETutf8mb4;说明field_type 存的是统一后的内部类型不是源库原始类型。is_primary_key 用于生成 DAG 任务的分区键判断。enable_llm 这个字段值得说道——它控制该字段是否进入 LLM 提示词手机号、身份证这类敏感字段置成 FALSE既防止数据泄露也减少 token 消耗。元数据同步时要特别注意源库表结构变更后跑一次同步任务更新 dataset_field但用户手工填的 field_comment 不能被覆盖。我见过不止一个系统在同步时把注释冲掉最后数据字典变成一堆空注释LLM 问答全靠猜字段含义效果惨不忍睹。同步逻辑应该是字段新增则插入字段删除则标记废弃字段类型变更则更新 field_type 但保留原注释。到这一步数据源的地基就打好了连接池管连接适配器管读写统一模型管类型元数据表管字典。下一章在这套地基之上看分布式 Pandas 引擎怎么把 TB 级数据跑起来。3. 分布式 Pandas 引擎TB 级数据处理的落地路径3.1 为什么计算底座选 Pandas 而不是纯 SQL数据源统一了接下来是“怎么算”。很多团队的第一反应是写 SQL但纯 SQL 在复杂清洗场景下很痛正则替换、自定义函数、跨库关联、机器学习预处理写 SQL 不是不能做是不好维护。Pandas 的表达力强Python 开发者上手快生态里现成的函数做窗口、分组、字符串处理都比 SQL 直观这是它被选作计算底座的核心原因。但单机 Pandas 在 TB 级数据面前有个硬伤内存装不下。常见解法是分布式的 Pandas 兼容框架最主流的是 Dask 的 dataframeAPI 几乎照着 pandas 抄另一个路线是 Modin Ray。如果团队想完全自己掌控分片逻辑也可以按「分区 合并」的思路自研这块我在下一小节展开。先看 Dask 的最小用法import dask.dataframe as dd df dd.read_csv( data/raw/2024*.csv, blocksize64MB, dtype{amount: float32, region: category}, ) result df.groupby(region).amount.sum().compute()代码说明dd.read_csv 用通配符把多文件读成一个 Dask DataFrameblocksize 决定每个分区大约 64MB。dtype 提前指定避免 Dask 在推断类型时因为某行脏数据把整列搞成 object。compute() 是触发实际计算的方法——在它之前所有操作只是构建任务图这也是 Dask 和 pandas 最不一样的地方。参数说明blocksize 别设太小分区太多会让调度器在任务编排上花掉大量时间也别太大超过单 worker 可用内存就直接 OOM。64MB 是多数场景的起点之后按数据行宽和 worker 内存调整。如果你自己实现分布式 pandas维护的核心任务之一就是让每个分片的大小落在 256MB 到 1GB 这个区间。3.2 数据分区与分片策略让 TB 级数据并行起来不用 Dask 的话自研分布式 pandas 的核心就是把数据切成互不依赖的分片。互不依赖是关键——如果分片之间有跨片关联比如全局去重、全局排序你就得再做一层 shuffle复杂度立刻上去。最省事的切法是按主键范围等宽切适合有明显自增主键的事实表def make_partitions(total_min: int, total_max: int, step: int 5_000_000): 按主键范围切分返回若干 (start, end) 区间。 parts [] start total_min while start total_max: end min(start step - 1, total_max) parts.append((start, end)) start end 1 return parts参数说明step 是每个分片的最大主键跨度我习惯按行数设 500 万行宽比较大几十个字段时降到 100 万。分片太大会导致单个 worker 内存吃紧分片太小则任务数爆炸调度开销超过计算收益。有了分片区间每个 worker 的任务就是一段独立查询加一段独立 pandas 处理for start, end in make_partitions(1, total_id, step5_000_000): query fSELECT * FROM orders WHERE id BETWEEN {start} AND {end} df pd.DataFrame(adapter.read(query)) result analyze(df) result.to_parquet( fstage/orders_{start}_{end}.parquet, enginepyarrow, compressionsnappy, )这个循环的妙处在于中间结果直接落盘。任何一个分片失败只需要找到对应的 parquet 文件重跑那一个区间不需要从头再来。这是 TB 级数据处理里最实用的后悔药。逻辑说明adapter.read 就是上一章 DataSourceAdapter 里的 read 方法包一层 pd.DataFrame 转成 pandas 对象analyze 是你自己的业务函数to_parquet 写入 stage 目录。所有分片完成后再起一个合并任务把 stage 下的 parquet 读进来做最终聚合。如果最终聚合也有压力就再分一层——先按天聚再按周聚典型的增量聚合套路。3.3 内存和算力的关键参数一组能直接抄的配置Pandas 处理 TB 级数据真正吃内存的是中间结果。这里列一组常用的参数按重要性排参数推荐值适用场景blocksize64MBDask 分区大小控制任务粒度npartitionsCPU 核数 × 24Dask 分区数不是越大越好float64 → float32开启数值列内存直接减半精度一般够用object → category低基数列开启枚举值少的列内存可降 510 倍parquet 压缩snappy压缩率与解压速度的平衡点表格里最容易忽略的是 category 这一行。很多人在 pandas 里不建 category导致一列只有十几个枚举值的“省份”字段占掉几百 MB。改成 category 后 groupby 也更快因为底层走的是整数编码。示例代码如下def read_optimized(path: str) - pd.DataFrame: df pd.read_parquet(path, columns[region, amount, is_valid]) df[amount] df[amount].astype(float32) df[region] df[region].astype(category) df[is_valid] df[is_valid].astype(bool) return df代码说明read_optimized 读 parquet 时只选需要的列减少磁盘 IO读完后立刻做类型转换。float32 在求和聚合场景下误差累积可以接受但如果指标是金额且需要精确到分这行别照抄回到统一模型那章说过的 Decimal 方案。内存策略里有个反直觉的点分区数不是越多越快。当分片小到一定程度磁盘 IO 和任务调度的开销会超过并行计算收益表现在监控上就是 CPU 使用率不高、任务倒是特别多。理想状态是所有 worker 的 CPU 保持 70% 以上如果你的系统跑完一轮发现大量时间花在等待上先看分片数是不是过多了。4. LLM 智能问答、低代码集成与 DAG 调度三件事的串法4.1 LLM 问答链路元数据、提示词与查询生成LLM 智能问答在这类系统里的定位是让业务人员用自然语言查数而不是写 SQL。链路不复杂用户提问 → 后端从元数据服务拉取相关表的数据字典 → 拼进提示词 → 让 LLM 生成 pandas 代码 → 在受限环境执行 → 把结果渲染成表格或图表。这条链路上最容易翻车的是提示词。数据字典如果不给全模型就只能靠猜给全了又可能超出上下文窗口。常见做法是先用轻量检索从元数据里筛出和问题最相关的字段只把这些字段拼进 schema_text。具体用哪个 LLM 框架反而不是重点只要兼容 OpenAI 接口后面换模型只需要改一个配置项。build_query_prompt 的模板大致是这样def build_query_prompt(user_question: str, schema_text: str) - str: return f你是数据分析助手。请根据数据字典生成 pandas 代码。 要求 1. 只使用 schema_text 中存在的字段名不要自己发明列名 2. 字段含义不明确时输出「无法确定」不要猜测 3. 只输出一个 python 代码块不要附带解释 数据字典 {schema_text} 用户问题{user_question} 代码说明把「只能用存在的字段」和「不确定就直说」写进提示词是控制幻觉最便宜的手段。生成代码后不能直接 exec——要放进受限命名空间去掉文件写和网络相关的 builtins只暴露 pandas 和指定的数据路径。参数说明LLM 调用时 temperature 固定为 0max_tokens 给 512 到 1024 之间。temperature 不为 0 的话同一问题两次查询可能生成不同的字段名业务侧会非常困惑。schema_text 里字段注释的质量直接决定问答准确率这就是上一章元数据表里要把 field_comment 当一等公民对待的原因。这里多说一句LLM 生成的结果也不是直接信。可以再加一层结果校验用 LLM as judge 的思路让模型判断生成结果和原始问题是否匹配。虽然多一次调用延迟但对核心指标查询来说是值得的。4.2 低代码集成拖拽节点与参数透传低代码集成的本质是「把 pandas 算子变成可拖拽的积木」。前端拖拽节点、连线构成图后端拿到的是一个 JSON 图然后按图执行。节点类型不用太多几类就够读取数据源、过滤、聚合、字段计算、关联、写目标表。一份节点图长这样node_graph { nodes: [ {id: read_1, type: read_datasource, params: {datasource_id: 12, table: orders}}, {id: filter_1, type: filter, params: {column: amount, op: gt, value: 1000}}, {id: agg_1, type: groupby_agg, params: {by: [region], agg: {amount: sum}}}, {id: write_1, type: write_datasink, params: {target_table: region_amount}} ], edges: [[read_1, filter_1], [filter_1, agg_1], [agg_1, write_1]] }执行器拿到这份 JSON 后先做拓扑排序这里和 DAG 调度是同一套逻辑然后逐个节点调用对应算子。低代码节点跟 DAG 任务的区别是DAG 的节点是「整个数据处理流程」低代码的节点是「一步 pandas 操作」。实际系统里一个低代码图可以被封装成 DAG 里的一个节点这就是两级嵌套的编排模型。参数说明edges 数组用 [上游, 下游] 表示依赖。前端连线时一定要防止两个节点直接成环但更可靠的是后端在保存时做一次环检测。前端校验只是体验层面的提醒后端校验才是底线。4.3 Python 后端与 Vue3 前端从 REST 到 WebSocketVue3 前端在这里的角色是数据目录、低代码画布、DAG 编排页面和问答对话界面。同步接口用 FastAPI 的 REST 没毛病但任务进度推送不能靠前端轮询。轮询会让后端在任务多的时候被无意义请求淹没所以进度推送要上 WebSocket。后端实现一个最小的进度广播中心from fastapi import WebSocket class ProgressHub: def __init__(self): self.connections: set[WebSocket] set() async def connect(self, ws: WebSocket): await ws.accept() self.connections.add(ws) async def broadcast(self, task_id: str, progress: int, msg: str): for ws in list(self.connections): try: await ws.send_json( {task_id: task_id, progress: progress, msg: msg} ) except RuntimeError: self.connections.discard(ws)逻辑说明broadcast 遍历所有连接推送进度。except RuntimeError 是处理「浏览器端已经断开但服务端还保留着连接」的情况——如果不清掉下一次广播会一直报错甚至导致任务回调失败。这个细节属于那种不遇到线上事故不会注意到的点。Vue3 侧的逻辑相对简单const ws new WebSocket(ws://${location.host}/ws/progress) ws.onmessage (e) { const data JSON.parse(e.data) progressMap.value[data.task_id] data.progress }说明progressMap 是一个 reactive 对象Vue3 的响应式系统会直接驱动进度条更新。连接断开后要做重连可以补一个心跳重连的封装但核心逻辑就这么几行。这里提一句 2026 年做 Vue3 后台管理系统的通用现状状态管理用 Pinia、UI 库选 Element Plus 或 Ant Design Vue 的较多重点是长列表渲染和 WebSocket 状态管理这两块做好系统的基本盘就稳了。4.4 DAG 工作流调度任务编排的核心实践DAG 调度解决的是「任务按依赖关系自动排队执行」的问题。核心概念有三个节点任务、边依赖、运行实例一次具体的执行。调度器只负责判断哪些任务可以被触发执行器拿到触发指令后调用对应算子或子工作流。拓扑排序是 DAG 调度的心脏。标准 Kahn 算法实现如下from collections import deque def topo_run(tasks: dict[str, list[str]], on_run): tasks 形如 {task_a: [task_b], ...}表示 a 依赖 b。 adj {t: [] for t in tasks} indegree {t: 0 for t in tasks} for t, deps in tasks.items(): indegree[t] len(deps) for dep in deps: adj.setdefault(dep, []).append(t) ready deque([t for t, d in indegree.items() if d 0]) while ready: t ready.popleft() on_run(t) for nxt in adj.get(t, []): indegree[nxt] - 1 if indegree[nxt] 0: ready.append(nxt)代码说明indegree 记录每个任务还有多少个依赖没完成初始入度为 0 的任务进 ready 队列。每完成一个任务下游任务的 indegree 减一减到 0 就说明它的依赖齐了可以执行。on_run 是你自己的执行回调可以是跑 Python 脚本、触发低代码图、或调用另一个子 DAG。实际生产里要在这层之上再加两层一是任务实例表记录每个任务每次运行的开始时间、结束时间、状态、日志路径失败重试也记录二是调度策略cron 表达式或事件触发。参数上要注意调度周期用 cron 表达式重试次数从 0 到 3退避间隔用 60/300/900 秒这种阶梯式失败后立刻重试通常还是失败等一会儿反而能恢复。5. 避坑与常见问题排查跑通到跑稳的距离功能跑通只需要半天跑稳需要几周。这章写的是我在类似系统上踩过或者帮别人排查时常见的坑每条都是现象、原因、解决三件套。这些坑不一定全部只在 ezdata 里出现但只要你按这个方向搭系统大概率会遇到。5.1 数据源连接池耗尽导致任务假死现象数据源任务卡在「读取中」日志没有任何报错偶尔出现 TimeoutError重启服务后正常一阵子又卡住。原因适配器每次 read 都新建连接用完没有归还几十个并发任务把数据库的 max_connections 打满。后续任务全部排队等连接看起来就像「假死」。解决把连接管理收口到 PooledDB 连接池同时在任务执行前先探活探活失败直接失败退出不要排队。每次任务结束在 finally 里执行 conn.close() 归还连接。加了探活之后至少问题会以「数据源不可用」的形式快速暴露而不是卡成一锅粥。5.2 Pandas 分区数据倾斜导致 OOM现象一批 10 个分片9 个在 5 分钟内跑完最后一个跑了半小时然后被 OOM Killer 杀掉整批任务失败要全部重来。原因按主键范围等宽分片适合均匀分布的自增主键但业务表往往不均匀——某些大客户的订单量是普通用户的几百倍集中在一小段 id 区间等宽分片就变成了「9 个小任务 1 个大任务」。解决分片键改成 hash 分桶把用户维度的数据打散或者在分片之前先做一次采样统计按数据量动态切分。hash 分桶代码很简单def hash_partition(user_id: str, n: int) - int: 按 user_id 哈希分桶适合用户维度事务表。 return hash(str(user_id)) % n说明每个桶里的数据量会均匀很多代价是如果分析任务本身需要按时间范围过滤hash 分桶会破坏时间连续性。这种场景可以先按时间粗筛再按用户 hash 做二级分区折中处理。5.3 LLM 问答幻觉答非所问怎么收敛现象用户问「上季度华北销售额」模型一本正经返回华南数据还带精确到小数点后两位的金额。业务方直接截图发给领导场面很难看。原因元数据字段注释缺失或含糊模型只能靠字段名猜。比如表里同时存在 sale_amount 和 order_amount没有注释说明区别模型选错的概率很高。解决第一字段注释作为硬指标元数据里没写注释的字段不参与 LLM 问答第二提示词里明确「字段含义不明确就回答无法确定」第三生成代码后做列名校验发现代码里引用了不存在的列就重新生成一次而不是直接执行。三层下来幻觉能压掉一大半但不可能归零——LLM 本身就是概率模型这是做问答功能时要有的预期。5.4 DAG 循环依赖检测不生效现象保存工作流时报「存在循环依赖」但图上看起来明明是合法的 DAG或者保存时没报错运行时任务一直不触发。原因前端只做了连线层面的校验后端保存接口没有做环检测导致库里存了带环的图。节点 ID 变更后旧引用没清理也会出现类似问题。解决把环检测放到服务端保存前统一跑一次 DFS。代码很短def has_cycle(tasks: dict[str, list[str]]) - bool: state: dict[str, int] {} def dfs(node: str) - bool: state[node] 1 for nxt in tasks.get(node, []): if state.get(nxt) 1: return True if state.get(nxt) is None and dfs(nxt): return True state[node] 2 return False return any(state.get(n) is None and dfs(n) for n in tasks)代码说明state 用 0/1/2 表示未访问、访问中、完成。dfs 时遇到状态为 1 的节点说明回到了路径上的点即存在环。tasks.get(node, []) 的写法是为了兼容那些被引用但没定义的节点——这种情况也应该在保存时直接报错。前端的校验只是体验后端的校验才是边界。5.5 前端一次性渲染十万行页面卡死现象查询结果 10 万行Vue3 页面白屏或卡到无法滚动浏览器直接弹「无响应」。原因表格组件按 v-for 把所有行都渲染成真实 DOM10 万行就是几十万个节点浏览器扛不住。解决用支持虚拟滚动的表格组件比如 vxe-table或者后端默认限制查询 1 万行超过就提示用户缩小范围或走导出。虚拟滚动的核心思路是只渲染可视区域内的行滚动时动态替换。对内部数据平台来说先限制查询行数是最省事的兜底虚拟滚动是体验优化两个一起做最好。6. 把 ezdata 真正用起来验收链路与一个值得做的优化最后这一章不讲新概念只讲落地和验证方法。先给一条最短验收链路准备一个 MySQL 实例建两张表 orders 和 region各灌 100 万行测试数据。然后按这个顺序走配置数据源 → 跑一次元数据同步 → 在低代码画布里拖一个「读取 orders → 按 region 聚合 → 写表」的流程 → 手动运行一次确认结果正确 → 再把这个流程挂到一个每天 2 点的 DAG 任务上让调度器自动触发。这四步走通系统主链路就通了。验证的节奏建议是先用 100 万行跑通逻辑再翻到 1000 万行看内存和耗时最后拿一个真实业务表压一次观察 worker 的 CPU 曲线是否保持在 60% 以上。如果 CPU 上不去先怀疑分片过大或过小而不是怀疑并行框架。值得做的优化是给 LLM 问答加一层「数据字典缓存与裁剪」。我见过不少系统每次问答都去读全量元数据拼进 prompttoken 消耗大、响应也慢。正确做法是把元数据按 dataset 维度缓存起来用户提问后先用关键词召回相关 dataset只把命中的字段注释拼进 schema_text。这一步能把 prompt 体积缩小一个数量级问答延迟和成本都明显下降。我的习惯是每周在测试环境做一次全链路演练删掉所有任务实例清空中间结果目录然后从元数据同步开始重新跑一遍 DAG确认没有任何隐藏的手工依赖。刚开始做会觉得很麻烦坚持下来之后系统发布新版本就再也不用担心「上周那个任务是不是还依赖着某台机器上的某个文件」。希望帮到你。本文还有配套的精品资源点击获取