☰
t3code:第三类编码任务的约定式协作与工程化实践
2026/10/26 9:26:05 网站建设 项目流程

1. 从“t3code”这个名字说起:它到底是什么

第一次看到“t3code”这个词,很多人会下意识地把它当成某个开源库、某个命令行工具,或者某个内部项目的代号。我最初接触它的时候也是这么想的,翻了半天仓库和文档,结果发现它并不是一个能直接npm install或者pip install就完事的东西。它更像是一种编码协作的约定,一套围绕“第三类编码任务”组织起来的工作方法,核心是把那些介于“纯手工”和“全自动”之间的编码环节,用一种可复用、可交接、可追溯的方式固定下来。

说得再直白一点:t3code 解决的是“同一件事,不同人做出来结果不一样”的问题。比如数据清洗、字段映射、批量重命名、格式转换、接口字段对齐这类活儿,写脚本能做,手工也能做,但每个人写法不同、命名不同、边界处理不同,最后交接的时候就是一团乱麻。t3code 的思路是先把这类任务抽象成“第三类编码任务”——既不是一次性的临时脚本,也不是需要长期维护的生产级服务,而是有明确输入输出、有固定处理链路、需要被反复执行和交接的中间层任务。

它适合谁来参考?我个人的判断是三类人:第一类是经常要处理重复性编码任务但又不想每次都重写脚本的开发者;第二类是团队里负责“脏活累活”但希望把这些活儿标准化的人;第三类是想把自己零散的经验沉淀成可复用资产的技术人。如果你属于这三类中的任何一类,t3code 这套思路值得花时间琢磨。

我之所以愿意花篇幅来拆解它,是因为我在实际项目里踩过太多“临时脚本变祖传代码”的坑。一个当初只打算跑一次的数据迁移脚本,半年后还在被三个人改来改去,字段名对不上、日志看不懂、报错了不知道从哪查。t3code 的价值就在于,它逼你在动手之前先想清楚:这个任务的输入是什么、输出是什么、中间经过哪几步、每一步的失败边界在哪。想清楚这些,代码反而成了最不重要的部分。

2. 核心设计思路拆解:为什么要把编码任务分成三类

2.1 三类编码任务的划分逻辑

t3code 最核心的一个判断,是把编码任务按“生命周期”和“复用频率”分成三类。这个划分不是拍脑袋来的,而是从实际协作场景里总结出来的。

第一类是一次性任务,比如临时查个数据、验证一个想法、跑个一次性统计。这类任务的特点是生命周期极短,写完就扔,不需要考虑复用和交接。第二类是生产级任务,比如线上服务、定时调度、对外接口,这类任务需要长期维护,有明确的 SLA 和监控要求。第三类就是 t3code 关注的中间层任务——它比一次性任务活得久,比生产级任务轻,但恰恰是团队里最容易被忽视、最容易出问题的一类。

我举个实际例子你就明白了。假设你要把一批用户数据从旧系统迁移到新系统,字段名从user_name改成username,日期格式从20240101改成2024-01-01,还要过滤掉测试账号。这个任务可能只跑一次,但它涉及多个步骤、多个边界条件,而且跑完之后可能还要复查、可能要重跑、可能要交给别人核对。这就是典型的第三类任务。

为什么这类任务最容易出问题?因为它既没有一次性任务的“随意性豁免”,又没有生产级任务的“工程化约束”。大家默认它“随便写写就行”,结果就是每个人写出来的东西都不一样,出了问题只能找原作者,原作者走了就成了黑盒。

2.2 为什么选择“约定优于配置”的路线

t3code 在设计上走的是“约定优于配置”的路线,这一点我很认同。它没有搞一套复杂的配置文件,也没有强制你用某个框架,而是通过一套命名和结构约定,让不同人写出来的东西能互相看懂。

具体来说,它约定每个第三类任务都要有四个固定部分:输入声明、处理步骤、输出声明、失败处理。输入声明负责说清楚“我吃什么”,处理步骤负责说清楚“我怎么吃”,输出声明负责说清楚“我吐什么”,失败处理负责说清楚“吃坏了怎么办”。这四个部分用固定的目录结构和命名规则组织起来,任何人拿到一个任务目录,都能在几分钟内看懂它在干什么。

这个设计的好处是显而易见的。我试过在团队里推行这套约定,最直接的感受是:新人接手老任务的时间从平均半天缩短到半小时以内。因为结构是固定的,他不需要去猜“这个脚本的入口在哪”“那个中间文件是干嘛的”,按约定找就行了。

当然,约定优于配置也有代价。它牺牲了一部分灵活性,比如你想加一个自定义步骤,就得按它的规则来命名和放置。但在我看来,对于第三类任务这种“复用频率高但复杂度不高”的场景,灵活性带来的收益远小于一致性带来的收益。

2.3 与常见脚本管理方式的对比

很多人会问:这不就是写脚本吗?我建个scripts目录,按日期或者功能命名,不也一样?我一开始也这么想,但实际用下来发现差别很大。

普通脚本管理方式的问题在于“隐式知识太多”。比如你写了一个migrate.py,里面硬编码了数据库连接、字段映射、过滤规则。别人拿到这个脚本,得从头读一遍才知道它依赖什么、输出什么、边界在哪。而 t3code 的方式是强制把这些隐式知识显式化:输入是什么、输出是什么、中间经过哪几步、每步失败怎么处理,全部写在固定位置。

我做过一个对比,同样是“用户数据迁移”这个任务,用普通脚本方式,交接时需要口头说明大约 15 分钟,加上对方自己读代码 20 分钟,总共 35 分钟左右。用 t3code 方式,交接时只需要指一下目录结构,对方自己看 10 分钟就能上手。这个差距在任务数量多的时候会被放大得很明显。

另一个差别是可追溯性。普通脚本跑完就完了,最多留个日志文件。t3code 约定每个任务都要保留输入快照、输出快照和处理日志,这样出了问题可以快速定位是哪一步、哪个输入导致的。我踩过一次坑,一个数据清洗任务跑完之后发现少了几十条记录,因为没有输入快照,根本没法复现。后来按 t3code 的方式补了快照,同样的问题再出现时,五分钟就定位到了是某个过滤条件写错了。

3. 核心细节解析与实操要点

3.1 输入声明的写法与注意事项

输入声明是 t3code 任务的第一步,也是最容易被轻视的一步。很多人觉得“输入不就是个文件路径吗”,随便写写就行。但实际经验告诉我,输入声明写得好不好,直接决定了这个任务能不能被复用。

t3code 约定输入声明要包含四个要素:来源、格式、范围、校验规则。来源是说数据从哪来,是本地文件、数据库还是接口;格式是说数据的结构,是 CSV、JSON 还是其他;范围是说这次处理哪些数据,是全量还是增量;校验规则是说怎么判断输入是合法的,比如字段数量、类型、非空约束等。

我举个例子。假设你要处理一批订单数据,输入声明应该这么写:

input: source: local_file path: ./data/orders_20240101.csv format: csv schema: - name: order_id type: string required: true - name: amount type: float required: true - name: created_at type: datetime format: "%Y-%m-%d %H:%M:%S" range: all validation: - rule: row_count > 0 - rule: no_duplicate(order_id)

这么写的好处是,任何人拿到这个声明,不用看代码就知道输入长什么样、有什么约束。而且校验规则是自动执行的,输入不合法直接报错,不会等到处理到一半才发现问题。

注意:输入声明里的校验规则不要写得太复杂,够用就行。我见过有人把业务逻辑也写进校验规则里,结果校验比处理还慢,这就本末倒置了。

3.2 处理步骤的拆分原则

处理步骤是 t3code 任务的核心部分。这里的核心原则是:每一步只做一件事,每一步都有明确的输入和输出。

为什么强调“一步只做一件事”?因为第三类任务最容易出的问题就是“一步做太多”。比如一个步骤里既做了字段映射,又做了数据过滤,还做了格式转换。这样一旦出错,你根本不知道是哪部分出的问题。拆成三步之后,每步单独测试、单独验证,出错时一眼就能看出是哪步的问题。

t3code 约定处理步骤用有序列表的方式组织,每一步都要声明:步骤名称、输入、处理逻辑、输出、失败处理。我拿一个实际的字段映射任务来举例:

steps: - name: rename_fields input: raw_data logic: | 将 user_name 重命名为 username 将 user_email 重命名为 email output: renamed_data on_failure: abort - name: filter_test_accounts input: renamed_data logic: | 过滤掉 email 以 @test.com 结尾的记录 output: filtered_data on_failure: skip_and_log - name: format_dates input: filtered_data logic: | 将 created_at 从 YYYYMMDD 格式转换为 YYYY-MM-DD output: final_data on_failure: abort

这里有个细节值得说:on_failure的取值。t3code 约定了几种常见的失败处理策略,比如abort表示直接终止,skip_and_log表示跳过当前记录并记录日志,retry表示重试。选择哪种策略取决于这一步的重要程度。像字段重命名这种基础操作,失败了就应该直接终止;像过滤测试账号这种辅助操作,失败了可以跳过并记录,不影响主流程。

实操心得:我建议在写处理步骤之前,先用纸笔把步骤画出来,确认每一步的输入输出能串起来。我踩过的坑是,写着写着发现某一步的输出格式和下一步的输入格式对不上,又得回头改,很浪费时间。

3.3 输出声明与失败处理的配合

输出声明和失败处理是 t3code 里最容易被忽略、但实际价值最高的两个部分。很多人写脚本只关心“跑成功”,不关心“跑失败了怎么办”和“跑完了输出在哪”。结果就是出了问题只能重跑,重跑又可能产生重复数据。

t3code 约定输出声明要包含:输出位置、输出格式、输出校验、幂等性保证。输出位置和格式好理解,关键是输出校验和幂等性。输出校验是说跑完之后要检查输出是否符合预期,比如记录数是否在合理范围内、关键字段是否为空。幂等性是说同一个任务跑多次,结果应该是一样的,不会产生重复数据。

失败处理则要和输出声明配合。比如一个任务失败了,是保留中间结果还是清理掉?是记录失败原因还是直接抛异常?这些都要在声明里写清楚。我一般会约定:中间结果保留,最终输出清理。这样既方便排查问题,又不会污染最终结果。

output: path: ./output/orders_processed.csv format: csv validation: - rule: row_count > 0 - rule: no_null(username) idempotent: true failure_handling: keep_intermediate: true cleanup_final: true log_level: detailed

这套配合下来,最大的好处是可重跑。我遇到过很多次任务跑到一半失败的情况,如果没有幂等性保证,重跑就会产生重复数据,还得手工去重。有了这套约定之后,直接重跑就行,不用担心数据污染。

3.4 目录结构约定与命名规范

t3code 对目录结构和命名有明确的约定,这是保证一致性的基础。我把它总结成一个标准模板:

t3code_tasks/ task_name/ input.yaml steps.yaml output.yaml failure.yaml run.py logs/ snapshots/ input/ output/

每个任务一个目录,目录名用下划线命名,比如user_data_migration。四个 YAML 文件分别对应输入声明、处理步骤、输出声明、失败处理。run.py是执行入口,负责读取配置、按步骤执行、处理失败。logs放日志,snapshots放输入输出快照。

这个结构看起来简单,但实际用起来很顺手。我试过在一个有二十多个第三类任务的团队里推行这套结构,最大的感受是“找东西不用问人”。想知道某个任务吃什么、吐什么、怎么跑,直接进目录看四个 YAML 文件就行,不用去翻代码。

注意:run.py里不要写业务逻辑,业务逻辑全部放在steps.yaml里声明。run.py只负责调度和执行。这样做的目的是让业务逻辑和调度逻辑分离,改业务逻辑不用动代码,改调度逻辑不影响业务。

4. 实操过程与核心环节实现

4.1 从零搭建一个 t3code 任务的完整流程

光说理论没意思,我拿一个实际做过的任务来完整走一遍流程。任务背景是:把一批商品数据从旧格式转换成新格式,旧格式是 CSV,字段有product_id、product_name、price、category,新格式要求字段名改成id、name、price、category,价格单位从“分”改成“元”,并且过滤掉价格小于 0 的异常记录。

第一步,建目录。我在t3code_tasks下建了一个product_data_transform目录。

第二步,写输入声明。输入是本地 CSV 文件,字段和类型如下:

input: source: local_file path: ./data/products_raw.csv format: csv schema: - name: product_id type: string required: true - name: product_name type: string required: true - name: price type: integer required: true - name: category type: string required: false range: all validation: - rule: row_count > 0 - rule: no_duplicate(product_id)

第三步,写处理步骤。这里我拆成三步:字段重命名、价格单位转换、异常过滤。

steps: - name: rename_fields input: raw_data logic: | 将 product_id 重命名为 id 将 product_name 重命名为 name output: renamed_data on_failure: abort - name: convert_price_unit input: renamed_data logic: | 将 price 从分转换为元,即 price = price / 100 output: converted_data on_failure: abort - name: filter_invalid_price input: converted_data logic: | 过滤掉 price < 0 的记录 output: final_data on_failure: skip_and_log

第四步,写输出声明。输出到本地 CSV,要求幂等。

output: path: ./output/products_transformed.csv format: csv validation: - rule: row_count > 0 - rule: no_null(id) - rule: price >= 0 idempotent: true

第五步,写失败处理。中间结果保留,最终输出清理,日志详细。

failure_handling: keep_intermediate: true cleanup_final: true log_level: detailed

第六步,写run.py。这个文件只负责调度,不写业务逻辑。核心逻辑是:读输入声明、校验输入、按步骤执行、写输出、处理失败。

import yaml import pandas as pd import os import shutil def load_config(task_dir): with open(os.path.join(task_dir, 'input.yaml')) as f: input_cfg = yaml.safe_load(f) with open(os.path.join(task_dir, 'steps.yaml')) as f: steps_cfg = yaml.safe_load(f) with open(os.path.join(task_dir, 'output.yaml')) as f: output_cfg = yaml.safe_load(f) with open(os.path.join(task_dir, 'failure.yaml')) as f: failure_cfg = yaml.safe_load(f) return input_cfg, steps_cfg, output_cfg, failure_cfg def validate_input(df, input_cfg): for rule in input_cfg.get('validation', []): if rule == 'row_count > 0' and len(df) == 0: raise ValueError('Input is empty') if rule.startswith('no_duplicate'): field = rule.split('(')[1].rstrip(')') if df[field].duplicated().any(): raise ValueError(f'Duplicate found in {field}') def execute_steps(df, steps_cfg): for step in steps_cfg['steps']: try: if step['name'] == 'rename_fields': df = df.rename(columns={'product_id': 'id', 'product_name': 'name'}) elif step['name'] == 'convert_price_unit': df['price'] = df['price'] / 100 elif step['name'] == 'filter_invalid_price': df = df[df['price'] >= 0] except Exception as e: if step['on_failure'] == 'abort': raise elif step['on_failure'] == 'skip_and_log': print(f'Step {step["name"]} failed: {e}') return df def main(): task_dir = os.path.dirname(os.path.abspath(__file__)) input_cfg, steps_cfg, output_cfg, failure_cfg = load_config(task_dir) df = pd.read_csv(input_cfg['path']) validate_input(df, input_cfg) os.makedirs(os.path.join(task_dir, 'snapshots', 'input'), exist_ok=True) shutil.copy(input_cfg['path'], os.path.join(task_dir, 'snapshots', 'input', 'raw.csv')) result = execute_steps(df, steps_cfg) os.makedirs(os.path.dirname(output_cfg['path']), exist_ok=True) result.to_csv(output_cfg['path'], index=False) os.makedirs(os.path.join(task_dir, 'snapshots', 'output'), exist_ok=True) shutil.copy(output_cfg['path'], os.path.join(task_dir, 'snapshots', 'output', 'final.csv')) if __name__ == '__main__': main()

这套代码跑下来,整个流程是清晰可控的。输入校验不通过直接报错,处理步骤按顺序执行,输出有校验,失败有处理。我实测下来,同样的任务用这种方式写,第一次可能比随手写脚本多花二十分钟,但后续每次复用、每次交接、每次排查问题,省下来的时间远超这二十分钟。

4.2 参数选择与计算过程

在 t3code 任务里,参数选择是个容易被忽视但很关键的环节。我拿几个常见参数来说说我的选择逻辑。

第一个是批处理大小。如果任务涉及大量数据,一次性加载可能内存不够,需要分批处理。批大小的选择取决于可用内存和单条记录大小。我一般用这个公式估算:batch_size = available_memory * 0.6 / record_size。比如可用内存 4GB,单条记录平均 1KB,那批大小大约是4 * 1024 * 1024 * 0.6 / 1 ≈ 250万。但实际我会取小一点,比如 100万,留出余量。

第二个是重试次数和间隔。如果任务涉及外部依赖,比如读数据库、调接口,失败重试是必要的。我一般设置重试 3 次,间隔用指数退避,第一次 1 秒,第二次 2 秒,第三次 4 秒。这个配置在大多数场景下够用,不会因为重试太频繁把下游打挂,也不会因为间隔太长导致任务卡住。

第三个是日志级别。t3code 约定日志级别分三档:summary只记录开始、结束、结果统计;detailed记录每一步的输入输出和耗时;debug记录每条记录的处理情况。我一般默认用detailed,排查问题时临时切到debug。debug级别日志量很大,不要长期开着。

实操心得:批处理大小不要设成固定值,最好根据输入数据量动态调整。我试过固定批大小,结果小数据量时批太小效率低,大数据量时批太大内存爆。后来改成min(1000000, max(1000, total_rows // 10)),效果好很多。

4.3 实操现场记录:一次真实的任务执行

我拿最近做的一个日志清洗任务来记录一下实操过程。任务背景是:把一批应用日志从原始格式清洗成结构化格式,原始日志是每行一条,格式是时间戳 | 级别 | 模块 | 消息,要求输出 CSV,字段拆开,并且过滤掉DEBUG级别的日志。

我先建了log_cleaning目录,然后写输入声明。输入是多个日志文件,放在./data/logs/目录下,格式是文本,每行一条。校验规则是文件数量大于 0,每行至少包含 3 个分隔符。

input: source: local_directory path: ./data/logs/ format: text pattern: "*.log" validation: - rule: file_count > 0 - rule: line_separator_count >= 3

处理步骤拆成三步:解析行、过滤级别、格式化输出。

steps: - name: parse_lines input: raw_lines logic: | 按 | 分割每行,提取 timestamp, level, module, message output: parsed_records on_failure: skip_and_log - name: filter_debug input: parsed_records logic: | 过滤掉 level == DEBUG 的记录 output: filtered_records on_failure: abort - name: format_output input: filtered_records logic: | 将 timestamp 转换为 ISO 格式 将记录组装为 CSV 行 output: final_csv on_failure: abort

输出声明和失败处理按标准模板写。然后跑起来,第一次跑发现parse_lines步骤有大约 2% 的行解析失败,原因是消息里本身包含|符号,导致分割字段数不对。我在parse_lines的逻辑里加了一条:如果分割后字段数大于 4,把第 4 个及之后的字段合并为 message。改完之后重跑,解析失败率降到 0.1% 以下。

这个坑很典型:分隔符冲突。日志消息里包含分隔符是常见情况,处理方式要么是限制分割次数,要么是转义。我选择限制分割次数,因为日志格式相对固定,前三个字段不会有分隔符,从第四个字段开始合并就行。

4.4 性能优化与资源控制

t3code 任务虽然比生产级任务轻,但性能问题该考虑还是要考虑。我总结了几条实际有效的优化经验。

第一条是能流式处理就不要全量加载。如果任务只是逐行转换,不需要全局聚合,那就用流式读取,一行一行处理,内存占用恒定。我试过一个 10GB 的日志文件,全量加载需要 20GB 内存,流式处理只需要几十 MB。

第二条是能并行就不要串行。如果任务可以按文件或按数据块拆分,那就用多进程并行处理。Python 里用multiprocessing.Pool就行,注意控制并发数,一般设为 CPU 核数的 1.5 倍左右。

第三条是能缓存就不要重复计算。如果某个步骤的结果会被后续步骤多次使用,那就缓存起来。比如字段映射表,加载一次缓存起来,不要每次用的时候都重新读。

注意:并行处理时要注意输出顺序。如果任务对输出顺序有要求,并行处理后需要按原始顺序重新排序。我踩过这个坑,并行处理后输出顺序乱了,下游任务依赖顺序,结果出了错。

5. 常见问题与排查技巧实录

5.1 输入校验失败怎么办

输入校验失败是 t3code 任务最常见的报错。我整理了一个排查顺序,按这个顺序走,基本能覆盖 90% 的情况。

第一步,看校验规则本身有没有写错。比如no_duplicate(product_id)里的字段名是不是和实际字段名一致,row_count > 0是不是写成了row_count >= 0。我见过有人把字段名写错,结果校验一直失败,查了半天才发现是拼写问题。

第二步,看输入数据本身有没有问题。比如字段缺失、类型不对、有空值。这时候需要看输入快照,对比一下预期和实际。t3code 约定保留输入快照就是为了这个场景。

第三步,看校验规则的执行顺序。如果多个校验规则之间有依赖,比如先检查字段存在再检查字段唯一性,那顺序不能乱。我一般把基础校验放前面,业务校验放后面。

报错信息可能原因排查方法
row_count > 0 failed输入为空检查输入文件是否为空,路径是否正确
no_duplicate failed字段有重复值用去重统计确认重复数量和重复值
required field missing字段缺失检查输入 schema 和实际字段是否一致
type mismatch类型不对检查字段类型声明和实际值类型

5.2 处理步骤中途失败怎么恢复

处理步骤中途失败是另一个高频问题。t3code 的失败处理约定里,keep_intermediate: true就是为了应对这种情况。中间结果保留下来,可以从失败的那一步继续,不用从头重跑。

具体操作是:先看日志确认是哪一步失败、失败原因是什么。然后检查中间结果,确认失败步骤之前的输出是完整的。修复问题后,从失败步骤开始重跑。t3code 的run.py支持指定起始步骤,比如python run.py --start-from convert_price_unit。

这里有个细节要注意:中间结果的版本管理。如果中间结果被覆盖了,就没法恢复了。我一般约定中间结果按步骤名和时间戳命名,比如intermediate/rename_fields_20240101_120000.csv,这样每次跑都保留一份,不会互相覆盖。

实操心得:我建议在run.py里加一个--dry-run选项,只校验输入和输出,不实际执行处理步骤。这样可以在正式跑之前先确认配置没问题,避免跑到一半才发现配置写错了。

5.3 输出结果不符合预期怎么排查

输出结果不符合预期,可能是处理逻辑问题,也可能是输入问题,还可能是输出校验规则问题。我一般按“从后往前”的顺序排查。

先看输出校验规则有没有通过。如果校验规则本身写得太严,比如price >= 0但实际业务允许价格为负,那就会误报。如果校验规则没问题,再看输出数据本身。对比输出快照和预期结果,找出差异记录。

然后看处理步骤的逻辑。把差异记录单独拿出来,手工走一遍处理步骤,看是哪一步产生了差异。我一般会在run.py里加一个--trace选项,对指定记录打印每一步的中间结果,方便定位。

最后看输入数据。如果输入数据本身就有问题,比如字段值有隐藏字符、编码不对,那处理步骤再正确也没用。我踩过一次坑,输入 CSV 里有 BOM 头,导致第一个字段名多了几个不可见字符,字段映射一直失败。后来在输入声明里加了encoding: utf-8-sig才解决。

5.4 任务复用时的常见坑

t3code 任务设计出来就是为了复用的,但复用的时候有几个坑要注意。

第一个坑是硬编码路径。如果任务里写死了输入输出路径,换个环境就跑不了。我一般约定路径用相对路径,相对于任务目录,这样整个任务目录拷贝到哪都能跑。

第二个坑是依赖版本不一致。如果任务依赖某个库的特定版本,换环境后版本不对,行为可能不一样。我一般会在任务目录里放一个requirements.txt,明确依赖版本。

第三个坑是环境变量缺失。如果任务依赖环境变量,比如数据库连接串,换环境后环境变量没设置,任务就失败。我一般约定环境变量在input.yaml里声明,run.py启动时检查,缺失就报错。

复用场景常见坑解决方案
换机器路径硬编码用相对路径
换环境依赖版本不一致固定依赖版本
换用户环境变量缺失声明并检查环境变量
换数据输入格式变化输入校验规则覆盖新格式

5.5 独家避坑技巧汇总

最后分享几条我从实际踩坑中总结出来的技巧,都是文档里不会写的。

第一条,输入快照要压缩。如果输入数据很大,快照会占很多磁盘。我一般用 gzip 压缩,读取的时候自动解压。这样既保留了快照,又不会把磁盘撑爆。

第二条,日志要分文件。不要把所有日志写到一个文件里,按任务、按日期分文件。我一般用logs/{task_name}_{date}.log的命名方式,方便查找和清理。

第三条,失败处理要分级。不是所有失败都需要终止任务。我一般把失败分成三级:critical直接终止,warning跳过并记录,info只记录不跳过。这样可以在保证任务不中断的前提下,尽可能多地处理数据。

第四条,定期清理中间结果。中间结果保留是为了排查问题,但保留太久会占磁盘。我一般约定保留最近 7 天的中间结果,超过的自动清理。清理脚本放在任务目录里,定期跑一次。

第五条,任务目录要版本控制。t3code 任务目录里的 YAML 文件和run.py都应该纳入版本控制,这样每次修改都有记录,出问题可以回滚。我一般用 Git 管理,每个任务一个分支或者一个子目录。

这套东西用下来,我最大的感受是:第三类任务的价值不在于写得多快,而在于写完之后别人能不能看懂、能不能复用、出了问题能不能查。t3code 这套约定,本质上是在用工程化的思路管理那些“看起来不值得工程化”的任务。实际用下来,投入产出比很高,尤其是团队协作场景下,省下来的沟通成本和排查成本远超预期。

需要专业的网站建设服务?

联系我们获取免费的网站建设咨询和方案报价,让我们帮助您实现业务目标

立即咨询