上周我花了两天时间试图把一个在本地跑得挺顺的脚本改造成一个能稳定处理上千条数据的自动化流程。结果脚本是跑起来了但过程堪称“练废了”——日志混乱、内存泄漏、异常中断后数据对不上号最后不得不回滚到最原始的手动状态从头梳理。这让我想起一个老生常谈却又总被我们下意识忽略的问题把一个能“跑起来”的脚本变成一个能“用起来”的系统中间隔着的不是代码量而是一整套工程化思维的鸿沟。我们太容易沉浸在“功能实现”的兴奋里却忘了问自己如果这个脚本要跑一百遍、一千遍如果中途网络波动、如果输入文件格式有误、如果磁盘空间不足……它还能不能优雅地处理并告诉我到底发生了什么今天我不想分享一个成功的案例恰恰相反我想完整复盘这次“练废了”的经历。记录下从“单次跑通”到“批量翻车”的全过程把那些看似微不足道、却足以让整个流程崩溃的细节摊开来讲。这不仅仅是一次失败记录更是一份从脚本思维转向工程思维的避坑检查清单。如果你也经常写一些“一次性”脚本却对把它们投入生产环境感到心虚那么这篇内容或许能帮你省下大量折腾的时间。1. 从“跑通就行”到“稳定运行”到底差在哪最开始我的目标非常单纯有一个本地脚本它能读取一个CSV文件对每一行数据调用某个在线API把返回的结果解析后写入一个新的文件。在测试环境下用一个10行的小文件跑一切顺利。输出正确控制台打印着“Success”的字样感觉良好。于是我自然而然地想“这不就成了吗把输入文件换成那个有5000行数据的生产文件不就行了” 这就是第一个也是最致命的思维陷阱用单次成功线性外推批量成功。两者的差异远不止是数据量乘以500倍那么简单。它们是完全不同的物种单次运行环境是干净的内存是初始状态所有注意力都集中在这一条数据上。任何错误都会立刻暴露你手动介入处理即可。批量运行状态是累积的内存、文件句柄、网络连接错误是偶发且叠加的你需要一套机制来自动处理异常、记录上下文、保证任务的可中断与可恢复。我当时没意识到我的脚本只具备了“单次运行”的能力却背负了“批量运行”的期望。它缺少了作为一个可靠系统所必需的几个核心支柱可观测性脚本在干什么进度如何遇到了什么问题不能只靠print。健壮性网络超时怎么办API返回了非预期格式怎么办磁盘满了怎么办可维护性三个月后我还看得懂当时的错误日志吗配置如API密钥、超时时间硬编码在脚本里怎么安全地修改可操作性任务跑到一半崩溃了是从头开始还是跳过已成功的部分继续如何手动终止或暂停一个运行中的任务忽略这些就像造了一辆能在自家后院平稳行驶的玩具车却直接把它开上了高速公路。翻车几乎是必然的。2. “练废”现场那些被忽略的细节如何连环引爆当我换上5000行的生产文件按下回车后灾难开始以各种意想不到的方式上演。2.1 日志之殇当print成为灾难放大器最初的脚本我用print来记录进度和错误。print(fProcessing row {i}: {data}) try: result call_api(data) print(fSuccess: {result}) except Exception as e: print(fError on row {i}: {e})在批量运行中这导致了两个问题信息淹没成千上万的print信息与标准错误流混在一起滚动飞快。当一个真正的错误出现时它瞬间就被刷屏淹没了我不得不费力地回溯终端历史。缺乏结构所有信息都是扁平化的文本。我无法轻松地过滤出“所有失败的行”也无法统计成功率。当需要向他人报告问题或自己事后分析时面对一坨文本无从下手。教训在批量任务中必须使用结构化的日志系统如Python的logging模块。至少区分INFO进度、WARNING可处理的异常、ERROR任务失败等级别并输出到文件。这样你可以用grep或日志分析工具快速定位问题。2.2 异常处理的“假动作”我的异常处理看起来写了但其实是“假动作”。try: response requests.get(url, timeout5) data response.json() except Exception as e: print(fAPI call failed: {e}) # 然后呢这条数据怎么处理任务继续吗我只是打印了错误然后脚本继续处理下一条数据。这带来了更严重的问题数据一致性被破坏失败的数据被静默跳过导致最终输出文件的记录数少于输入文件。如果没有严格的校验这种数据丢失很难被发现。错误原因被丢弃Exception太笼统了。是网络超时是认证失败还是服务器返回了500错误不同的错误需要不同的后续策略例如网络超时可以重试认证失败则应立即停止。教训异常处理必须是有策略的。需要细化异常类型捕获更具体的异常如requests.exceptions.Timeout,requests.exceptions.HTTPError,json.JSONDecodeError。决定失败策略是重试最多几次是跳过并记录到“失败清单”还是立即终止整个任务保证状态可追溯任何失败都必须将错误类型、错误详情、出错的数据标识如行号、ID完整地记录到日志或专门的错误报告中。2.3 资源泄漏沉默的“内存杀手”脚本在循环中调用API、处理数据、生成结果对象。我忘了在Python中大的中间变量如果不及时释放引用会一直占用内存。all_results [] # 危险 for row in big_data_list: processed heavy_processing(row) # 可能生成很大的对象 all_results.append(processed) # 继续使用 all_results...当处理几千条数据时all_results列表可能变得非常庞大导致内存耗尽MemoryError程序被系统杀死。更隐蔽的是文件句柄泄漏如果在循环内打开文件或网络连接而没有确保关闭系统资源会逐渐耗尽。教训对于批量任务要有“流式处理”或“分批次处理”的意识。及时释放对于不需要聚合所有结果的任务处理完一条就应写入输出文件然后丢弃该条数据在内存中的引用。使用生成器对于大的输入使用生成器yield而非一次性加载全部到内存。显式管理资源使用with语句上下文管理器来确保文件、网络连接等资源被正确关闭。监控资源在长时间运行的任务中可以定期打印或记录内存使用情况如psutil库以便早期发现问题。2.4 配置与硬编码藏在代码里的“地雷”API的URL、密钥、超时时间、数据库连接字符串……这些都被我直接写死在代码里。API_KEY sk-123456...abc # 直接暴露 BASE_URL https://api.example.com这带来了安全、维护和协作的多重问题安全风险如果代码被上传到GitHub等公开仓库密钥直接泄露。环境隔离困难开发、测试、生产环境通常使用不同的配置。硬编码意味着每次切换环境都要改代码极易出错。协作障碍队友拿到你的代码还需要问你密钥是什么或者自己去找地方修改。教训配置必须与代码分离。使用配置文件将配置写入config.ini,config.yaml或config.json文件并在.gitignore中忽略它们。使用环境变量这是更推荐的方式尤其适合密钥等敏感信息。通过操作系统环境变量或.env文件来设置。# .env 文件 API_KEYsk-123456...abc# 代码中读取 import os api_key os.getenv(API_KEY)提供配置模板在仓库中提供一个config.example.ini或.env.example文件说明各个配置项的含义方便他人使用。3. 从废墟中重建一个最小可行的工程化脚本框架经历了上述崩溃后我停下来不再盲目增加数据量而是着手重构。目标是建立一个最小可行但具备工程化特征的脚本框架。这个框架不一定复杂但必须解决上述核心痛点。3.1 第一步建立清晰的项目结构即使是单脚本项目一个清晰的结构也利于管理。my_data_pipeline/ ├── config/ # 配置目录 │ ├── settings.py # 读取环境变量的配置模块 │ └── .env # 本地环境变量被.gitignore ├── src/ # 源代码 │ └── processor.py # 主处理逻辑 ├── logs/ # 日志目录运行时生成 ├── outputs/ # 输出目录运行时生成 ├── inputs/ # 输入数据 ├── requirements.txt # 依赖列表 └── README.md # 项目说明如何设置和运行3.2 第二步实现配置管理config/settings.py:import os from dotenv import load_dotenv load_dotenv() # 从 .env 文件加载环境变量 class Config: API_KEY os.getenv(API_KEY) BASE_URL os.getenv(BASE_URL, https://api.default.com) REQUEST_TIMEOUT int(os.getenv(REQUEST_TIMEOUT, 30)) MAX_RETRIES int(os.getenv(MAX_RETRIES, 3)) LOG_LEVEL os.getenv(LOG_LEVEL, INFO) config Config().env文件API_KEYyour_secret_key_here BASE_URLhttps://api.production.com LOG_LEVELDEBUG3.3 第三步设置结构化日志在主脚本开头或一个专门的utils/logger.py中import logging import sys from pathlib import Path from config.settings import config def setup_logger(name): logger logging.getLogger(name) logger.setLevel(getattr(logging, config.LOG_LEVEL)) # 控制台处理器 console_handler logging.StreamHandler(sys.stdout) console_format logging.Formatter(%(asctime)s - %(name)s - %(levelname)s - %(message)s) console_handler.setFormatter(console_format) logger.addHandler(console_handler) # 文件处理器 log_dir Path(logs) log_dir.mkdir(exist_okTrue) file_handler logging.FileHandler(log_dir / f{name}.log, encodingutf-8) file_format logging.Formatter(%(asctime)s - %(name)s - %(levelname)s - %(module)s:%(lineno)d - %(message)s) file_handler.setFormatter(file_format) logger.addHandler(file_handler) return logger # 使用 logger setup_logger(__name__) logger.info(脚本启动) logger.error(处理第%s行时发生API错误, row_id, exc_infoTrue) # exc_infoTrue 会记录异常堆栈3.4 第四步设计健壮的核心处理循环这是重构的核心重点在于状态管理和错误处理策略。import time from typing import List, Dict, Any from config.settings import config # ... 导入自定义的logger和工具函数 def process_batch(input_data: List[Dict[str, Any]]) - Dict[str, Any]: 处理一批数据返回汇总结果。 results { total: len(input_data), success: 0, failed: 0, failures: [] # 记录详细的失败信息 } for idx, row in enumerate(input_data): row_id row.get(id, idx) logger.info(开始处理记录 ID: %s, row_id) retry_count 0 success False while retry_count config.MAX_RETRIES and not success: try: # 核心业务逻辑调用API response call_api_with_retry(row, config) processed_data parse_response(response) # 写入输出流式写入避免内存累积 write_to_output(processed_data) results[success] 1 success True logger.info(记录 ID: %s 处理成功, row_id) except TransientError as e: # 自定义的临时错误如网络超时 retry_count 1 if retry_count config.MAX_RETRIES: wait_time 2 ** retry_count # 指数退避 logger.warning(记录 ID: %s 遇到临时错误[%s]第%s次重试等待%ss, row_id, e, retry_count, wait_time) time.sleep(wait_time) else: logger.error(记录 ID: %s 重试%s次后仍失败: %s, row_id, config.MAX_RETRIES, e) results[failed] 1 results[failures].append({id: row_id, error: str(e), type: TransientError}) except BusinessError as e: # 自定义的业务错误如数据格式不对重试无意义 logger.error(记录 ID: %s 业务逻辑错误跳过: %s, row_id, e) results[failed] 1 results[failures].append({id: row_id, error: str(e), type: BusinessError}) break # 跳出重试循环 except Exception as e: # 兜底的未知错误记录并停止整个任务可能是更安全的选择 logger.critical(记录 ID: %s 遇到未预期错误停止处理: %s, row_id, e, exc_infoTrue) results[failed] 1 results[failures].append({id: row_id, error: str(e), type: UnexpectedError}) raise # 重新抛出终止任务 # 可选每处理N条或每隔一段时间记录一次进度并强制清理内存如gc.collect()) if idx % 100 0: logger.info(进度报告: 已处理 %s/%s 条 成功 %s 失败 %s, idx1, results[total], results[success], results[failed]) return results3.5 第五步主程序与优雅退出主程序负责组装整个流程读取配置、初始化、加载数据、分批次处理、汇总报告、清理资源。def main(): logger.info( 数据处理管道启动 ) start_time time.time() try: # 1. 加载输入数据考虑使用生成器或分块读取大文件 input_data load_input_data(inputs/data.csv) # 2. 分批次处理对于极大文件至关重要 batch_size 100 all_results {total: 0, success: 0, failed: 0, failures: []} for i in range(0, len(input_data), batch_size): batch input_data[i:ibatch_size] logger.info(开始处理批次 %s (条目 %s 到 %s), i//batch_size 1, i, ilen(batch)) batch_result process_batch(batch) # 合并结果 all_results[success] batch_result[success] all_results[failed] batch_result[failed] all_results[failures].extend(batch_result[failures]) all_results[total] len(input_data) # 3. 生成最终报告 logger.info( 处理完成 ) logger.info(总计: %s, 成功: %s, 失败: %s, all_results[total], all_results[success], all_results[failed]) if all_results[failures]: logger.warning(失败详情已写入日志及文件 failures.json) save_failures(all_results[failures]) except KeyboardInterrupt: logger.warning(用户中断执行。) except Exception as e: logger.critical(主程序发生致命错误: %s, e, exc_infoTrue) return 1 # 返回非零错误码 finally: # 确保资源清理的代码放在这里 logger.info(清理资源...) # ... 关闭文件、数据库连接等 elapsed time.time() - start_time logger.info(总运行时间: %.2f 秒, elapsed) return 0 if __name__ __main__: exit(main())4. 核心思维转变从写脚本到设计数据管道经过这次重构我得到的不仅仅是一个更健壮的脚本更重要的是一种思维模式的转变。我不再是写一个“脚本”而是在设计一个微型的“数据管道”。这意味着你需要像对待一个产品一样考虑它的全生命周期输入标准化数据从哪里来格式是否稳定是否需要预处理或校验处理原子化每个处理步骤是否职责单一是否易于测试和复用状态可追踪任何时刻你都能知道管道运行到哪一步成功了多少失败了多少失败的原因是什么。输出可验证输出数据的格式、数量、质量是否有机制验证失败可管理失败是常态。管道必须有明确的失败处理策略重试、跳过、告警、停止并提供清晰的失败报告。部署与调度这个管道如何被触发手动命令行定时任务cronCI/CD流水线它运行在什么环境下对于简单的个人任务可能不需要Airflow或Kafka这样重量级的工具。但即使只用纯Python脚本只要你具备了“管道思维”你写出来的代码也会自然而然地朝着可观测、可维护、可扩展的方向进化。你会开始习惯性地问自己如果这个环节出错了我该怎么知道知道了又该怎么处理这次“练废了”的经历价值远大于一次轻松的成功。它强迫我去正视那些在“跑通就行”阶段被忽略的工程细节。把这些细节补上虽然不会让脚本的“功能”增加一分但却能让它的“可靠性”发生质变。下次当你再写一个准备重复使用的脚本时不妨先问问自己我准备好应对批量、异常和长期运行了吗如果没有那么现在就是开始构建那套“隐形”支撑系统的最佳时机。
上周我花了两天时间试图把一个在本地跑得挺顺的脚本改造成一个能稳定处理上千条数据的自动化流程。结果脚本是跑起来了但过程堪称“练废了”——日志混乱、内存泄漏、异常中断后数据对不上号最后不得不回滚到最原始的手动状态从头梳理。这让我想起一个老生常谈却又总被我们下意识忽略的问题把一个能“跑起来”的脚本变成一个能“用起来”的系统中间隔着的不是代码量而是一整套工程化思维的鸿沟。我们太容易沉浸在“功能实现”的兴奋里却忘了问自己如果这个脚本要跑一百遍、一千遍如果中途网络波动、如果输入文件格式有误、如果磁盘空间不足……它还能不能优雅地处理并告诉我到底发生了什么今天我不想分享一个成功的案例恰恰相反我想完整复盘这次“练废了”的经历。记录下从“单次跑通”到“批量翻车”的全过程把那些看似微不足道、却足以让整个流程崩溃的细节摊开来讲。这不仅仅是一次失败记录更是一份从脚本思维转向工程思维的避坑检查清单。如果你也经常写一些“一次性”脚本却对把它们投入生产环境感到心虚那么这篇内容或许能帮你省下大量折腾的时间。1. 从“跑通就行”到“稳定运行”到底差在哪最开始我的目标非常单纯有一个本地脚本它能读取一个CSV文件对每一行数据调用某个在线API把返回的结果解析后写入一个新的文件。在测试环境下用一个10行的小文件跑一切顺利。输出正确控制台打印着“Success”的字样感觉良好。于是我自然而然地想“这不就成了吗把输入文件换成那个有5000行数据的生产文件不就行了” 这就是第一个也是最致命的思维陷阱用单次成功线性外推批量成功。两者的差异远不止是数据量乘以500倍那么简单。它们是完全不同的物种单次运行环境是干净的内存是初始状态所有注意力都集中在这一条数据上。任何错误都会立刻暴露你手动介入处理即可。批量运行状态是累积的内存、文件句柄、网络连接错误是偶发且叠加的你需要一套机制来自动处理异常、记录上下文、保证任务的可中断与可恢复。我当时没意识到我的脚本只具备了“单次运行”的能力却背负了“批量运行”的期望。它缺少了作为一个可靠系统所必需的几个核心支柱可观测性脚本在干什么进度如何遇到了什么问题不能只靠print。健壮性网络超时怎么办API返回了非预期格式怎么办磁盘满了怎么办可维护性三个月后我还看得懂当时的错误日志吗配置如API密钥、超时时间硬编码在脚本里怎么安全地修改可操作性任务跑到一半崩溃了是从头开始还是跳过已成功的部分继续如何手动终止或暂停一个运行中的任务忽略这些就像造了一辆能在自家后院平稳行驶的玩具车却直接把它开上了高速公路。翻车几乎是必然的。2. “练废”现场那些被忽略的细节如何连环引爆当我换上5000行的生产文件按下回车后灾难开始以各种意想不到的方式上演。2.1 日志之殇当print成为灾难放大器最初的脚本我用print来记录进度和错误。print(fProcessing row {i}: {data}) try: result call_api(data) print(fSuccess: {result}) except Exception as e: print(fError on row {i}: {e})在批量运行中这导致了两个问题信息淹没成千上万的print信息与标准错误流混在一起滚动飞快。当一个真正的错误出现时它瞬间就被刷屏淹没了我不得不费力地回溯终端历史。缺乏结构所有信息都是扁平化的文本。我无法轻松地过滤出“所有失败的行”也无法统计成功率。当需要向他人报告问题或自己事后分析时面对一坨文本无从下手。教训在批量任务中必须使用结构化的日志系统如Python的logging模块。至少区分INFO进度、WARNING可处理的异常、ERROR任务失败等级别并输出到文件。这样你可以用grep或日志分析工具快速定位问题。2.2 异常处理的“假动作”我的异常处理看起来写了但其实是“假动作”。try: response requests.get(url, timeout5) data response.json() except Exception as e: print(fAPI call failed: {e}) # 然后呢这条数据怎么处理任务继续吗我只是打印了错误然后脚本继续处理下一条数据。这带来了更严重的问题数据一致性被破坏失败的数据被静默跳过导致最终输出文件的记录数少于输入文件。如果没有严格的校验这种数据丢失很难被发现。错误原因被丢弃Exception太笼统了。是网络超时是认证失败还是服务器返回了500错误不同的错误需要不同的后续策略例如网络超时可以重试认证失败则应立即停止。教训异常处理必须是有策略的。需要细化异常类型捕获更具体的异常如requests.exceptions.Timeout,requests.exceptions.HTTPError,json.JSONDecodeError。决定失败策略是重试最多几次是跳过并记录到“失败清单”还是立即终止整个任务保证状态可追溯任何失败都必须将错误类型、错误详情、出错的数据标识如行号、ID完整地记录到日志或专门的错误报告中。2.3 资源泄漏沉默的“内存杀手”脚本在循环中调用API、处理数据、生成结果对象。我忘了在Python中大的中间变量如果不及时释放引用会一直占用内存。all_results [] # 危险 for row in big_data_list: processed heavy_processing(row) # 可能生成很大的对象 all_results.append(processed) # 继续使用 all_results...当处理几千条数据时all_results列表可能变得非常庞大导致内存耗尽MemoryError程序被系统杀死。更隐蔽的是文件句柄泄漏如果在循环内打开文件或网络连接而没有确保关闭系统资源会逐渐耗尽。教训对于批量任务要有“流式处理”或“分批次处理”的意识。及时释放对于不需要聚合所有结果的任务处理完一条就应写入输出文件然后丢弃该条数据在内存中的引用。使用生成器对于大的输入使用生成器yield而非一次性加载全部到内存。显式管理资源使用with语句上下文管理器来确保文件、网络连接等资源被正确关闭。监控资源在长时间运行的任务中可以定期打印或记录内存使用情况如psutil库以便早期发现问题。2.4 配置与硬编码藏在代码里的“地雷”API的URL、密钥、超时时间、数据库连接字符串……这些都被我直接写死在代码里。API_KEY sk-123456...abc # 直接暴露 BASE_URL https://api.example.com这带来了安全、维护和协作的多重问题安全风险如果代码被上传到GitHub等公开仓库密钥直接泄露。环境隔离困难开发、测试、生产环境通常使用不同的配置。硬编码意味着每次切换环境都要改代码极易出错。协作障碍队友拿到你的代码还需要问你密钥是什么或者自己去找地方修改。教训配置必须与代码分离。使用配置文件将配置写入config.ini,config.yaml或config.json文件并在.gitignore中忽略它们。使用环境变量这是更推荐的方式尤其适合密钥等敏感信息。通过操作系统环境变量或.env文件来设置。# .env 文件 API_KEYsk-123456...abc# 代码中读取 import os api_key os.getenv(API_KEY)提供配置模板在仓库中提供一个config.example.ini或.env.example文件说明各个配置项的含义方便他人使用。3. 从废墟中重建一个最小可行的工程化脚本框架经历了上述崩溃后我停下来不再盲目增加数据量而是着手重构。目标是建立一个最小可行但具备工程化特征的脚本框架。这个框架不一定复杂但必须解决上述核心痛点。3.1 第一步建立清晰的项目结构即使是单脚本项目一个清晰的结构也利于管理。my_data_pipeline/ ├── config/ # 配置目录 │ ├── settings.py # 读取环境变量的配置模块 │ └── .env # 本地环境变量被.gitignore ├── src/ # 源代码 │ └── processor.py # 主处理逻辑 ├── logs/ # 日志目录运行时生成 ├── outputs/ # 输出目录运行时生成 ├── inputs/ # 输入数据 ├── requirements.txt # 依赖列表 └── README.md # 项目说明如何设置和运行3.2 第二步实现配置管理config/settings.py:import os from dotenv import load_dotenv load_dotenv() # 从 .env 文件加载环境变量 class Config: API_KEY os.getenv(API_KEY) BASE_URL os.getenv(BASE_URL, https://api.default.com) REQUEST_TIMEOUT int(os.getenv(REQUEST_TIMEOUT, 30)) MAX_RETRIES int(os.getenv(MAX_RETRIES, 3)) LOG_LEVEL os.getenv(LOG_LEVEL, INFO) config Config().env文件API_KEYyour_secret_key_here BASE_URLhttps://api.production.com LOG_LEVELDEBUG3.3 第三步设置结构化日志在主脚本开头或一个专门的utils/logger.py中import logging import sys from pathlib import Path from config.settings import config def setup_logger(name): logger logging.getLogger(name) logger.setLevel(getattr(logging, config.LOG_LEVEL)) # 控制台处理器 console_handler logging.StreamHandler(sys.stdout) console_format logging.Formatter(%(asctime)s - %(name)s - %(levelname)s - %(message)s) console_handler.setFormatter(console_format) logger.addHandler(console_handler) # 文件处理器 log_dir Path(logs) log_dir.mkdir(exist_okTrue) file_handler logging.FileHandler(log_dir / f{name}.log, encodingutf-8) file_format logging.Formatter(%(asctime)s - %(name)s - %(levelname)s - %(module)s:%(lineno)d - %(message)s) file_handler.setFormatter(file_format) logger.addHandler(file_handler) return logger # 使用 logger setup_logger(__name__) logger.info(脚本启动) logger.error(处理第%s行时发生API错误, row_id, exc_infoTrue) # exc_infoTrue 会记录异常堆栈3.4 第四步设计健壮的核心处理循环这是重构的核心重点在于状态管理和错误处理策略。import time from typing import List, Dict, Any from config.settings import config # ... 导入自定义的logger和工具函数 def process_batch(input_data: List[Dict[str, Any]]) - Dict[str, Any]: 处理一批数据返回汇总结果。 results { total: len(input_data), success: 0, failed: 0, failures: [] # 记录详细的失败信息 } for idx, row in enumerate(input_data): row_id row.get(id, idx) logger.info(开始处理记录 ID: %s, row_id) retry_count 0 success False while retry_count config.MAX_RETRIES and not success: try: # 核心业务逻辑调用API response call_api_with_retry(row, config) processed_data parse_response(response) # 写入输出流式写入避免内存累积 write_to_output(processed_data) results[success] 1 success True logger.info(记录 ID: %s 处理成功, row_id) except TransientError as e: # 自定义的临时错误如网络超时 retry_count 1 if retry_count config.MAX_RETRIES: wait_time 2 ** retry_count # 指数退避 logger.warning(记录 ID: %s 遇到临时错误[%s]第%s次重试等待%ss, row_id, e, retry_count, wait_time) time.sleep(wait_time) else: logger.error(记录 ID: %s 重试%s次后仍失败: %s, row_id, config.MAX_RETRIES, e) results[failed] 1 results[failures].append({id: row_id, error: str(e), type: TransientError}) except BusinessError as e: # 自定义的业务错误如数据格式不对重试无意义 logger.error(记录 ID: %s 业务逻辑错误跳过: %s, row_id, e) results[failed] 1 results[failures].append({id: row_id, error: str(e), type: BusinessError}) break # 跳出重试循环 except Exception as e: # 兜底的未知错误记录并停止整个任务可能是更安全的选择 logger.critical(记录 ID: %s 遇到未预期错误停止处理: %s, row_id, e, exc_infoTrue) results[failed] 1 results[failures].append({id: row_id, error: str(e), type: UnexpectedError}) raise # 重新抛出终止任务 # 可选每处理N条或每隔一段时间记录一次进度并强制清理内存如gc.collect()) if idx % 100 0: logger.info(进度报告: 已处理 %s/%s 条 成功 %s 失败 %s, idx1, results[total], results[success], results[failed]) return results3.5 第五步主程序与优雅退出主程序负责组装整个流程读取配置、初始化、加载数据、分批次处理、汇总报告、清理资源。def main(): logger.info( 数据处理管道启动 ) start_time time.time() try: # 1. 加载输入数据考虑使用生成器或分块读取大文件 input_data load_input_data(inputs/data.csv) # 2. 分批次处理对于极大文件至关重要 batch_size 100 all_results {total: 0, success: 0, failed: 0, failures: []} for i in range(0, len(input_data), batch_size): batch input_data[i:ibatch_size] logger.info(开始处理批次 %s (条目 %s 到 %s), i//batch_size 1, i, ilen(batch)) batch_result process_batch(batch) # 合并结果 all_results[success] batch_result[success] all_results[failed] batch_result[failed] all_results[failures].extend(batch_result[failures]) all_results[total] len(input_data) # 3. 生成最终报告 logger.info( 处理完成 ) logger.info(总计: %s, 成功: %s, 失败: %s, all_results[total], all_results[success], all_results[failed]) if all_results[failures]: logger.warning(失败详情已写入日志及文件 failures.json) save_failures(all_results[failures]) except KeyboardInterrupt: logger.warning(用户中断执行。) except Exception as e: logger.critical(主程序发生致命错误: %s, e, exc_infoTrue) return 1 # 返回非零错误码 finally: # 确保资源清理的代码放在这里 logger.info(清理资源...) # ... 关闭文件、数据库连接等 elapsed time.time() - start_time logger.info(总运行时间: %.2f 秒, elapsed) return 0 if __name__ __main__: exit(main())4. 核心思维转变从写脚本到设计数据管道经过这次重构我得到的不仅仅是一个更健壮的脚本更重要的是一种思维模式的转变。我不再是写一个“脚本”而是在设计一个微型的“数据管道”。这意味着你需要像对待一个产品一样考虑它的全生命周期输入标准化数据从哪里来格式是否稳定是否需要预处理或校验处理原子化每个处理步骤是否职责单一是否易于测试和复用状态可追踪任何时刻你都能知道管道运行到哪一步成功了多少失败了多少失败的原因是什么。输出可验证输出数据的格式、数量、质量是否有机制验证失败可管理失败是常态。管道必须有明确的失败处理策略重试、跳过、告警、停止并提供清晰的失败报告。部署与调度这个管道如何被触发手动命令行定时任务cronCI/CD流水线它运行在什么环境下对于简单的个人任务可能不需要Airflow或Kafka这样重量级的工具。但即使只用纯Python脚本只要你具备了“管道思维”你写出来的代码也会自然而然地朝着可观测、可维护、可扩展的方向进化。你会开始习惯性地问自己如果这个环节出错了我该怎么知道知道了又该怎么处理这次“练废了”的经历价值远大于一次轻松的成功。它强迫我去正视那些在“跑通就行”阶段被忽略的工程细节。把这些细节补上虽然不会让脚本的“功能”增加一分但却能让它的“可靠性”发生质变。下次当你再写一个准备重复使用的脚本时不妨先问问自己我准备好应对批量、异常和长期运行了吗如果没有那么现在就是开始构建那套“隐形”支撑系统的最佳时机。
平方复杂度无法被常规优化消除｜TaoToken 统一 Key 通道实测)
