1. 为什么我要从零搭一个轻量 AI 数据平台去年年底我手上同时跑着三个跟 AI 相关的内部小项目一个做文本清洗和标注一个做向量化入库还有一个做模型评测的数据回流。三个项目各写各的脚本各连各的数据库各用各的调度方式。最夸张的时候我一天要手动跑七次 Python 脚本中间任何一步挂了就得从头再来。那段时间我最大的感受不是技术难而是乱——数据在哪儿、跑到哪一步了、哪份数据是干净的、哪份是脏的全靠脑子记和翻聊天记录。后来我下定决心花了两周多的时间从 0 到 1 搭了一个轻量 AI 数据平台的初版。它不是什么大厂级别的数据中台也没有上百个微服务核心目标就三个把数据流转的链路固定下来、把每一步的执行状态记录下来、把重复的手工操作自动化掉。说白了就是让我自己少加班、少背锅、少在半夜被叫起来重跑任务。这篇文章我会把整个搭建过程拆开讲架构是怎么定的、为什么这么定、ETL 环节怎么设计、OperatorRegistry 这个核心组件怎么落地、踩了哪些坑、以及初版和生产环境之间到底差在哪。如果你也是一个人或者小团队在维护 AI 相关的数据链路或者你正准备从脚本堆里爬出来做一点工程化这篇内容应该能帮你少走一些弯路。我会尽量说人话把每个决策背后的“为什么”讲清楚而不是甩一堆名词让你自己猜。2. 架构决策轻量不等于随便边界要先划清楚2.1 先想清楚“不做什么”比想做什么更重要一开始我也想过要不要直接上一套成熟的开源调度平台比如 Airflow 或者 Dagster。但冷静下来评估了一下我的数据量不大日增数据在百万行级别任务数量也就几十个团队就我一个人维护。Airflow 功能确实全但它的部署成本、学习成本、以及出问题时的排查成本对一个人来说太重了。我需要的是能跑、能看、能改而不是能扩展、能集群、能多租户。所以我给自己划了三条边界不追求高可用初版允许单点故障挂了就重启先把链路跑通。不追求多租户只服务我自己和身边两三个同事不做权限体系。不追求实时批处理为主分钟级延迟完全能接受。这三条边界一划架构瞬间就清晰了。很多人在搭平台时容易陷入“一步到位”的陷阱结果功能还没用上维护成本先把自己压垮了。我的经验是初版的目标是验证链路不是验证架构的扩展性。2.2 整体分层四层结构各司其职最终我定下来的架构是四层层级职责我的选型接入层接收任务触发、参数传入FastAPI 简单 CLI调度层任务编排、依赖管理、状态跟踪自研轻量调度器 数据库状态表执行层具体 ETL 逻辑、算子执行OperatorRegistry 插件式算子存储层元数据、任务状态、中间结果PostgreSQL 本地文件系统这里重点说两个决策。第一为什么调度层要自研而不是用现成的。因为我的任务依赖关系很简单基本是线性的最多有个分支合并。自研一个基于数据库轮询的调度器代码量不到 500 行出问题我能直接看代码定位不用去翻别人的源码。第二为什么执行层要用 OperatorRegistry。这是整个平台的核心抽象后面会单独展开讲。2.3 存储选型PostgreSQL 扛下了所有元数据和状态存储我选了 PostgreSQL原因很直接我需要事务、需要 JSON 字段、需要可靠的并发控制。任务状态表用行锁就能解决并发更新问题中间结果用 JSONB 存查询和更新都很方便。本地文件系统用来存大块的中间数据比如清洗后的文本文件、向量文件数据库里只存路径和校验值。提示初版不要急着上对象存储本地文件系统在单机场景下性能更好调试也更方便。等数据量真的上来了再迁移迁移成本并不高。2.4 一个被忽略的决策日志和可观测性我一开始觉得日志不重要结果第一次任务失败时我花了两个小时才定位到是一个编码问题。从那以后我给每个算子都加了结构化日志记录输入行数、输出行数、耗时、异常堆栈。这些日志直接写进数据库的 task_log 表配合一个简单的查询页面排查效率提升了不止十倍。可观测性不是生产环境的专利初版就需要。3. 核心细节解析ETL 链路和 OperatorRegistry 怎么落地3.1 ETL 在 AI 数据平台里的特殊性传统 ETL 处理的是结构化数据字段固定、类型明确。但 AI 场景下的数据往往是半结构化甚至非结构化的文本、图片路径、JSON 嵌套、向量数组。这就导致两个问题一是schema 不稳定今天多一个字段明天少一个字段二是处理逻辑重清洗、分词、向量化、去重每一步都可能很耗时。我的应对策略是把 ETL 拆成算子每个算子只做一件事算子之间通过标准化的数据契约传递。数据契约就是一个 Python dict必须包含data、meta、stats三个 key。data是实际数据meta是元信息来源、时间戳、版本stats是统计信息行数、耗时、异常数。这样每个算子只需要关心自己的输入输出不用管上下游是谁。3.2 OperatorRegistry 的设计思路OperatorRegistry 说白了就是一个算子注册表。我用 Python 的装饰器实现核心代码大概长这样class OperatorRegistry: def __init__(self): self._operators {} def register(self, name): def decorator(cls): self._operators[name] cls return cls return decorator def get(self, name): if name not in self._operators: raise ValueError(fOperator {name} not found) return self._operators[name] registry OperatorRegistry() registry.register(clean_text) class CleanTextOperator: def run(self, input_data): # 具体清洗逻辑 pass这么设计的好处有三个。第一新增算子不需要改调度代码只要注册进去就能用。第二算子可以独立测试输入输出都是标准契约单测很好写。第三配置驱动任务配置里只写算子名字和参数调度器动态加载。3.3 算子之间的数据契约数据契约我定了一个基类所有算子必须继承class DataContract: def __init__(self, data, metaNone, statsNone): self.data data self.meta meta or {} self.stats stats or {rows_in: 0, rows_out: 0, errors: 0}这里有个坑我踩过一开始我没强制要求stats结果某个算子偷偷吞了异常导致下游数据量对不上排查了半天。后来我把stats变成必填并且调度器会在每个算子执行后校验rows_in和rows_out的差异超过阈值就告警。数据质量校验要前置不要等下游发现问题再回溯。3.4 任务配置的格式任务配置我用 YAML因为可读性好改起来方便。一个典型配置长这样task_name: daily_text_pipeline schedule: 0 2 * * * operators: - name: load_raw params: source: /data/raw/text - name: clean_text params: remove_html: true min_length: 10 - name: dedup params: method: simhash threshold: 0.9 - name: save_clean params: target: /data/clean/text调度器读取这个配置按顺序执行算子每一步的状态都写进数据库。如果某一步失败整个任务标记为失败并且记录失败算子和异常信息。3.5 状态管理和重试机制任务状态我设计了六种pending、running、success、failed、retrying、skipped。状态流转全部通过数据库事务保证一致性。重试机制我做了简单的指数退避最多重试三次。这里有个细节重试时要保证幂等否则会出现数据重复。我的做法是每个算子在写入前先检查目标是否存在存在就覆盖或者跳过具体策略由算子自己决定。注意不要无脑重试所有异常。像编码错误、文件不存在这种重试多少次都没用应该直接失败并告警。只有网络抖动、临时锁冲突这类才值得重试。4. 实操过程从零到跑通第一条链路4.1 环境准备和依赖安装我的运行环境是 Ubuntu 22.04Python 3.10。依赖不多核心就几个pip install fastapi uvicorn sqlalchemy psycopg2-binary pyyaml python-dateutil数据库用 Docker 起了一个 PostgreSQLdocker run -d --name ai-platform-db \ -e POSTGRES_PASSWORDyourpassword \ -e POSTGRES_DBai_platform \ -p 5432:5432 \ postgres:15这里提醒一句密码不要硬编码在代码里我用环境变量读取本地开发用.env文件生产环境用系统环境变量。4.2 数据库表结构设计核心表就四张表名用途关键字段task_definition任务定义task_name, config_yaml, scheduletask_run任务运行实例run_id, task_name, status, start_time, end_timeoperator_run算子运行记录run_id, operator_name, status, stats_jsontask_log日志run_id, level, message, timestamptask_run和operator_run是一对多关系通过run_id关联。stats_json用 JSONB 类型存算子的输入输出统计。4.3 调度器核心逻辑调度器我用了一个简单的轮询循环每 30 秒检查一次有没有到期的任务。核心逻辑def scheduler_loop(): while True: due_tasks find_due_tasks() for task in due_tasks: run_id create_task_run(task) try: execute_task(task, run_id) mark_success(run_id) except Exception as e: mark_failed(run_id, str(e)) time.sleep(30)execute_task里按顺序执行算子每个算子执行前后都更新operator_run表。这里有个优化点算子执行可以并行的地方就并行比如多个独立的清洗任务我用concurrent.futures开了线程池整体耗时降了 40% 左右。4.4 第一个算子的实现我写的第一个算子是load_raw从本地目录读取文本文件。实现很简单但有几个细节要注意registry.register(load_raw) class LoadRawOperator: def run(self, contract, params): source params[source] files glob.glob(f{source}/*.txt) data [] for f in files: with open(f, r, encodingutf-8) as fp: data.append({file: f, content: fp.read()}) contract.data data contract.stats[rows_out] len(data) return contract坑点编码问题。我一开始没指定encodingutf-8结果遇到 GBK 文件直接报错。后来我加了一个编码探测逻辑先尝试 UTF-8失败再尝试 GBK再失败就记录到错误文件里跳过。4.5 跑通第一条完整链路配置好任务后我手动触发了一次python cli.py run --task daily_text_pipeline执行过程大概是这样调度器创建task_run状态running。执行load_raw读取 1200 个文件rows_out1200。执行clean_text去掉 HTML 标签、过滤短文本rows_out980。执行dedupSimHash 去重rows_out870。执行save_clean写入目标目录。任务状态更新为success。整个链路跑下来大概 3 分钟比我之前手动跑快了不止一倍而且中间每一步都有记录出问题能直接定位。4.6 监控和告警的初版实现监控我做得比较简单一个 FastAPI 接口返回最近 24 小时的任务状态统计。告警用了一个 webhook任务失败时发一条消息到群里。这里的关键是告警要带上下文不能只发一句“任务失败”要带上 run_id、失败算子、异常信息这样收到消息就能直接开始排查。5. 踩过的坑和常见问题排查5.1 数据库连接池耗尽第一次跑批量任务时我开了 20 个线程并行执行算子结果数据库连接池直接爆了。原因是每个线程都去拿连接连接池默认只有 5 个。解决办法是限制并发数并且给每个线程复用连接。后来我把并发数控制在 8 个连接池调到 20问题解决。5.2 算子执行时间过长导致任务卡死有个算子做向量化单条数据要调模型1200 条数据跑了两个小时。调度器没有超时机制任务一直卡在running状态。后来我加了算子级超时超过配置时间就强制终止并标记失败。超时时间按算子配置默认 10 分钟。5.3 数据契约被意外修改有个算子直接修改了contract.data的结构导致下游算子拿到的数据格式不对。排查了很久才发现是上游算子的问题。后来我加了一条规则算子只能修改contract.data的内容不能改变它的类型。并且加了一个 schema 校验每个算子执行后检查数据类型是否符合预期。5.4 常见问题速查表问题现象可能原因排查方法解决方案任务一直 running算子卡死或超时查 operator_run 表加超时机制手动终止数据量对不上算子吞异常查 stats 差异强制 stats 校验数据库连接报错连接池耗尽查并发数和连接数限制并发调大连接池编码错误文件编码不统一查错误日志加编码探测重试后数据重复算子不幂等查目标数据写入前检查存在性5.5 几个独家避坑技巧日志要带 run_id不然多个任务并行时日志混在一起根本没法看。配置要版本化任务配置改了要记录版本不然出问题不知道是哪版配置导致的。算子要能单独跑我写了一个 CLI 命令可以单独执行某个算子方便调试。数据库要定期备份初版也要备份我吃过一次亏状态表被误删任务历史全没了。6. 初版和生产的真实差距6.1 差距一没有真正的分布式初版是单机跑所有算子都在一个进程里。生产环境需要分布式执行算子要能调度到不同机器上。这涉及到任务队列、资源调度、结果回传复杂度直接上一个台阶。我的建议是初版先把单机跑稳分布式等有真实需求再上。6.2 差距二没有完善的权限和审计初版没有用户体系谁都能触发任务、改配置。生产环境需要权限控制、操作审计、配置审批。这块我暂时没做因为团队小信任成本低。但如果要开放给更多人用这是必须补的。6.3 差距三监控和告警太粗糙初版的监控只有一个状态统计接口告警只有失败通知。生产环境需要更细的指标算子耗时分布、数据量趋势、异常率、资源使用率。告警也要分级不同级别走不同渠道。6.4 差距四没有数据质量体系初版的数据质量校验是零散的每个算子自己检查。生产环境需要统一的数据质量规则引擎支持配置化校验、自动阻断、质量报告。这块我打算下一步做先把规则抽象出来。6.5 差距五部署和升级太手工初版部署就是git pull然后重启进程升级靠手动。生产环境需要 CI/CD、灰度发布、回滚机制。这块我暂时用脚本凑合但心里清楚这不是长久之计。6.6 我的个人体会搭这个平台最大的收获不是技术上的而是思维方式的变化。以前写脚本想的是“怎么把这件事做完”现在做平台想的是“怎么让这件事可重复、可观测、可维护”。这个转变让我在后面的项目里少踩了很多坑。如果让我重新来一次我会在初版就加上三样东西算子超时、数据契约校验、配置版本化。这三样成本不高但能省下大量排查时间。另外不要追求一步到位初版的目标是跑通链路不是做完美架构。先让它跑起来再让它跑得稳最后才让它跑得快。这个平台后续我还会继续迭代下一步计划是把数据质量规则引擎做出来再把分布式执行加上。如果你也在做类似的事情欢迎交流踩过的坑可以一起填。
1. 为什么我要从零搭一个轻量 AI 数据平台去年年底我手上同时跑着三个跟 AI 相关的内部小项目一个做文本清洗和标注一个做向量化入库还有一个做模型评测的数据回流。三个项目各写各的脚本各连各的数据库各用各的调度方式。最夸张的时候我一天要手动跑七次 Python 脚本中间任何一步挂了就得从头再来。那段时间我最大的感受不是技术难而是乱——数据在哪儿、跑到哪一步了、哪份数据是干净的、哪份是脏的全靠脑子记和翻聊天记录。后来我下定决心花了两周多的时间从 0 到 1 搭了一个轻量 AI 数据平台的初版。它不是什么大厂级别的数据中台也没有上百个微服务核心目标就三个把数据流转的链路固定下来、把每一步的执行状态记录下来、把重复的手工操作自动化掉。说白了就是让我自己少加班、少背锅、少在半夜被叫起来重跑任务。这篇文章我会把整个搭建过程拆开讲架构是怎么定的、为什么这么定、ETL 环节怎么设计、OperatorRegistry 这个核心组件怎么落地、踩了哪些坑、以及初版和生产环境之间到底差在哪。如果你也是一个人或者小团队在维护 AI 相关的数据链路或者你正准备从脚本堆里爬出来做一点工程化这篇内容应该能帮你少走一些弯路。我会尽量说人话把每个决策背后的“为什么”讲清楚而不是甩一堆名词让你自己猜。2. 架构决策轻量不等于随便边界要先划清楚2.1 先想清楚“不做什么”比想做什么更重要一开始我也想过要不要直接上一套成熟的开源调度平台比如 Airflow 或者 Dagster。但冷静下来评估了一下我的数据量不大日增数据在百万行级别任务数量也就几十个团队就我一个人维护。Airflow 功能确实全但它的部署成本、学习成本、以及出问题时的排查成本对一个人来说太重了。我需要的是能跑、能看、能改而不是能扩展、能集群、能多租户。所以我给自己划了三条边界不追求高可用初版允许单点故障挂了就重启先把链路跑通。不追求多租户只服务我自己和身边两三个同事不做权限体系。不追求实时批处理为主分钟级延迟完全能接受。这三条边界一划架构瞬间就清晰了。很多人在搭平台时容易陷入“一步到位”的陷阱结果功能还没用上维护成本先把自己压垮了。我的经验是初版的目标是验证链路不是验证架构的扩展性。2.2 整体分层四层结构各司其职最终我定下来的架构是四层层级职责我的选型接入层接收任务触发、参数传入FastAPI 简单 CLI调度层任务编排、依赖管理、状态跟踪自研轻量调度器 数据库状态表执行层具体 ETL 逻辑、算子执行OperatorRegistry 插件式算子存储层元数据、任务状态、中间结果PostgreSQL 本地文件系统这里重点说两个决策。第一为什么调度层要自研而不是用现成的。因为我的任务依赖关系很简单基本是线性的最多有个分支合并。自研一个基于数据库轮询的调度器代码量不到 500 行出问题我能直接看代码定位不用去翻别人的源码。第二为什么执行层要用 OperatorRegistry。这是整个平台的核心抽象后面会单独展开讲。2.3 存储选型PostgreSQL 扛下了所有元数据和状态存储我选了 PostgreSQL原因很直接我需要事务、需要 JSON 字段、需要可靠的并发控制。任务状态表用行锁就能解决并发更新问题中间结果用 JSONB 存查询和更新都很方便。本地文件系统用来存大块的中间数据比如清洗后的文本文件、向量文件数据库里只存路径和校验值。提示初版不要急着上对象存储本地文件系统在单机场景下性能更好调试也更方便。等数据量真的上来了再迁移迁移成本并不高。2.4 一个被忽略的决策日志和可观测性我一开始觉得日志不重要结果第一次任务失败时我花了两个小时才定位到是一个编码问题。从那以后我给每个算子都加了结构化日志记录输入行数、输出行数、耗时、异常堆栈。这些日志直接写进数据库的 task_log 表配合一个简单的查询页面排查效率提升了不止十倍。可观测性不是生产环境的专利初版就需要。3. 核心细节解析ETL 链路和 OperatorRegistry 怎么落地3.1 ETL 在 AI 数据平台里的特殊性传统 ETL 处理的是结构化数据字段固定、类型明确。但 AI 场景下的数据往往是半结构化甚至非结构化的文本、图片路径、JSON 嵌套、向量数组。这就导致两个问题一是schema 不稳定今天多一个字段明天少一个字段二是处理逻辑重清洗、分词、向量化、去重每一步都可能很耗时。我的应对策略是把 ETL 拆成算子每个算子只做一件事算子之间通过标准化的数据契约传递。数据契约就是一个 Python dict必须包含data、meta、stats三个 key。data是实际数据meta是元信息来源、时间戳、版本stats是统计信息行数、耗时、异常数。这样每个算子只需要关心自己的输入输出不用管上下游是谁。3.2 OperatorRegistry 的设计思路OperatorRegistry 说白了就是一个算子注册表。我用 Python 的装饰器实现核心代码大概长这样class OperatorRegistry: def __init__(self): self._operators {} def register(self, name): def decorator(cls): self._operators[name] cls return cls return decorator def get(self, name): if name not in self._operators: raise ValueError(fOperator {name} not found) return self._operators[name] registry OperatorRegistry() registry.register(clean_text) class CleanTextOperator: def run(self, input_data): # 具体清洗逻辑 pass这么设计的好处有三个。第一新增算子不需要改调度代码只要注册进去就能用。第二算子可以独立测试输入输出都是标准契约单测很好写。第三配置驱动任务配置里只写算子名字和参数调度器动态加载。3.3 算子之间的数据契约数据契约我定了一个基类所有算子必须继承class DataContract: def __init__(self, data, metaNone, statsNone): self.data data self.meta meta or {} self.stats stats or {rows_in: 0, rows_out: 0, errors: 0}这里有个坑我踩过一开始我没强制要求stats结果某个算子偷偷吞了异常导致下游数据量对不上排查了半天。后来我把stats变成必填并且调度器会在每个算子执行后校验rows_in和rows_out的差异超过阈值就告警。数据质量校验要前置不要等下游发现问题再回溯。3.4 任务配置的格式任务配置我用 YAML因为可读性好改起来方便。一个典型配置长这样task_name: daily_text_pipeline schedule: 0 2 * * * operators: - name: load_raw params: source: /data/raw/text - name: clean_text params: remove_html: true min_length: 10 - name: dedup params: method: simhash threshold: 0.9 - name: save_clean params: target: /data/clean/text调度器读取这个配置按顺序执行算子每一步的状态都写进数据库。如果某一步失败整个任务标记为失败并且记录失败算子和异常信息。3.5 状态管理和重试机制任务状态我设计了六种pending、running、success、failed、retrying、skipped。状态流转全部通过数据库事务保证一致性。重试机制我做了简单的指数退避最多重试三次。这里有个细节重试时要保证幂等否则会出现数据重复。我的做法是每个算子在写入前先检查目标是否存在存在就覆盖或者跳过具体策略由算子自己决定。注意不要无脑重试所有异常。像编码错误、文件不存在这种重试多少次都没用应该直接失败并告警。只有网络抖动、临时锁冲突这类才值得重试。4. 实操过程从零到跑通第一条链路4.1 环境准备和依赖安装我的运行环境是 Ubuntu 22.04Python 3.10。依赖不多核心就几个pip install fastapi uvicorn sqlalchemy psycopg2-binary pyyaml python-dateutil数据库用 Docker 起了一个 PostgreSQLdocker run -d --name ai-platform-db \ -e POSTGRES_PASSWORDyourpassword \ -e POSTGRES_DBai_platform \ -p 5432:5432 \ postgres:15这里提醒一句密码不要硬编码在代码里我用环境变量读取本地开发用.env文件生产环境用系统环境变量。4.2 数据库表结构设计核心表就四张表名用途关键字段task_definition任务定义task_name, config_yaml, scheduletask_run任务运行实例run_id, task_name, status, start_time, end_timeoperator_run算子运行记录run_id, operator_name, status, stats_jsontask_log日志run_id, level, message, timestamptask_run和operator_run是一对多关系通过run_id关联。stats_json用 JSONB 类型存算子的输入输出统计。4.3 调度器核心逻辑调度器我用了一个简单的轮询循环每 30 秒检查一次有没有到期的任务。核心逻辑def scheduler_loop(): while True: due_tasks find_due_tasks() for task in due_tasks: run_id create_task_run(task) try: execute_task(task, run_id) mark_success(run_id) except Exception as e: mark_failed(run_id, str(e)) time.sleep(30)execute_task里按顺序执行算子每个算子执行前后都更新operator_run表。这里有个优化点算子执行可以并行的地方就并行比如多个独立的清洗任务我用concurrent.futures开了线程池整体耗时降了 40% 左右。4.4 第一个算子的实现我写的第一个算子是load_raw从本地目录读取文本文件。实现很简单但有几个细节要注意registry.register(load_raw) class LoadRawOperator: def run(self, contract, params): source params[source] files glob.glob(f{source}/*.txt) data [] for f in files: with open(f, r, encodingutf-8) as fp: data.append({file: f, content: fp.read()}) contract.data data contract.stats[rows_out] len(data) return contract坑点编码问题。我一开始没指定encodingutf-8结果遇到 GBK 文件直接报错。后来我加了一个编码探测逻辑先尝试 UTF-8失败再尝试 GBK再失败就记录到错误文件里跳过。4.5 跑通第一条完整链路配置好任务后我手动触发了一次python cli.py run --task daily_text_pipeline执行过程大概是这样调度器创建task_run状态running。执行load_raw读取 1200 个文件rows_out1200。执行clean_text去掉 HTML 标签、过滤短文本rows_out980。执行dedupSimHash 去重rows_out870。执行save_clean写入目标目录。任务状态更新为success。整个链路跑下来大概 3 分钟比我之前手动跑快了不止一倍而且中间每一步都有记录出问题能直接定位。4.6 监控和告警的初版实现监控我做得比较简单一个 FastAPI 接口返回最近 24 小时的任务状态统计。告警用了一个 webhook任务失败时发一条消息到群里。这里的关键是告警要带上下文不能只发一句“任务失败”要带上 run_id、失败算子、异常信息这样收到消息就能直接开始排查。5. 踩过的坑和常见问题排查5.1 数据库连接池耗尽第一次跑批量任务时我开了 20 个线程并行执行算子结果数据库连接池直接爆了。原因是每个线程都去拿连接连接池默认只有 5 个。解决办法是限制并发数并且给每个线程复用连接。后来我把并发数控制在 8 个连接池调到 20问题解决。5.2 算子执行时间过长导致任务卡死有个算子做向量化单条数据要调模型1200 条数据跑了两个小时。调度器没有超时机制任务一直卡在running状态。后来我加了算子级超时超过配置时间就强制终止并标记失败。超时时间按算子配置默认 10 分钟。5.3 数据契约被意外修改有个算子直接修改了contract.data的结构导致下游算子拿到的数据格式不对。排查了很久才发现是上游算子的问题。后来我加了一条规则算子只能修改contract.data的内容不能改变它的类型。并且加了一个 schema 校验每个算子执行后检查数据类型是否符合预期。5.4 常见问题速查表问题现象可能原因排查方法解决方案任务一直 running算子卡死或超时查 operator_run 表加超时机制手动终止数据量对不上算子吞异常查 stats 差异强制 stats 校验数据库连接报错连接池耗尽查并发数和连接数限制并发调大连接池编码错误文件编码不统一查错误日志加编码探测重试后数据重复算子不幂等查目标数据写入前检查存在性5.5 几个独家避坑技巧日志要带 run_id不然多个任务并行时日志混在一起根本没法看。配置要版本化任务配置改了要记录版本不然出问题不知道是哪版配置导致的。算子要能单独跑我写了一个 CLI 命令可以单独执行某个算子方便调试。数据库要定期备份初版也要备份我吃过一次亏状态表被误删任务历史全没了。6. 初版和生产的真实差距6.1 差距一没有真正的分布式初版是单机跑所有算子都在一个进程里。生产环境需要分布式执行算子要能调度到不同机器上。这涉及到任务队列、资源调度、结果回传复杂度直接上一个台阶。我的建议是初版先把单机跑稳分布式等有真实需求再上。6.2 差距二没有完善的权限和审计初版没有用户体系谁都能触发任务、改配置。生产环境需要权限控制、操作审计、配置审批。这块我暂时没做因为团队小信任成本低。但如果要开放给更多人用这是必须补的。6.3 差距三监控和告警太粗糙初版的监控只有一个状态统计接口告警只有失败通知。生产环境需要更细的指标算子耗时分布、数据量趋势、异常率、资源使用率。告警也要分级不同级别走不同渠道。6.4 差距四没有数据质量体系初版的数据质量校验是零散的每个算子自己检查。生产环境需要统一的数据质量规则引擎支持配置化校验、自动阻断、质量报告。这块我打算下一步做先把规则抽象出来。6.5 差距五部署和升级太手工初版部署就是git pull然后重启进程升级靠手动。生产环境需要 CI/CD、灰度发布、回滚机制。这块我暂时用脚本凑合但心里清楚这不是长久之计。6.6 我的个人体会搭这个平台最大的收获不是技术上的而是思维方式的变化。以前写脚本想的是“怎么把这件事做完”现在做平台想的是“怎么让这件事可重复、可观测、可维护”。这个转变让我在后面的项目里少踩了很多坑。如果让我重新来一次我会在初版就加上三样东西算子超时、数据契约校验、配置版本化。这三样成本不高但能省下大量排查时间。另外不要追求一步到位初版的目标是跑通链路不是做完美架构。先让它跑起来再让它跑得稳最后才让它跑得快。这个平台后续我还会继续迭代下一步计划是把数据质量规则引擎做出来再把分布式执行加上。如果你也在做类似的事情欢迎交流踩过的坑可以一起填。
