1. 项目概述为什么一个“数据 workflow 框架”值得你花两小时认真读完我第一次在客户现场看到 Metaflow 被用起来是在一家做实时风控的金融科技公司。他们原本用 Airflow 调度 Python 脚本但模型训练环节总卡在“环境不一致”上——开发机跑通的代码一上生产就报ModuleNotFoundError: No module named xgboost数据科学家本地调参用的是 pandas 1.5而调度集群里装的是 1.3结果.assign()行为突变线上预测指标一夜掉 3 个点。他们试过 Docker 封装、Conda 环境导出、甚至手写 requirements.txt 版本锁全都不够稳。直到团队里一个刚毕业的工程师悄悄把整个训练 pipeline 改成了 Metaflow三天后他们第一次实现了“本地调试 → 一键提交 → 生产运行”三步零修改。不是靠运气是 Metaflow 把“代码、数据、环境、计算资源”这四样东西用一套统一的抽象捆在了一起。这就是 Metaflow 的核心价值它不只解决“任务怎么调度”更解决“任务在哪、用什么、跑成什么样全程可追溯、可复现、可协作”。它不是另一个 Airflow 或 Prefect 的平替而是专为数据科学工作流data science workflows设计的框架——这里的“数据科学”特指那些包含 Jupyter 探索、模型训练、特征工程、A/B 测试验证、模型部署前评估等混合型任务的长链条作业。它默认支持 Python 原生语法不强制你写 YAML 配置它把每个 step 的输入输出自动版本化连中间 DataFrame 的 schema 变更都能被检测到它原生集成 AWS Step Functions 和 Batch也支持本地执行和 Kubernetes切换只需改一行配置。关键词就是可复现性reproducibility、可伸缩性scalability、开发者友好developer experience。如果你是数据科学家、ML 工程师或者正在从“单机脚本”向“可交付 pipeline”转型的分析师这篇教程不是“学个新工具”而是帮你把过去三个月反复重跑、反复 debug、反复解释“为什么结果不一样”的时间一次性收回来。2. 核心设计逻辑与选型依据为什么 Metaflow 不是“又一个 workflow 引擎”2.1 它解决的不是“调度问题”而是“数据科学协作熵增问题”先说结论Metaflow 的设计哲学本质是对抗数据科学工作流中的“熵”——即混乱、不可控、不可追溯的状态。传统 workflow 引擎如 Airflow的核心假设是“任务是确定性的、幂等的、输入输出明确的黑盒”。但数据科学任务恰恰相反一个train_model()函数可能依赖随机种子、GPU 显存状态、pandas 版本、甚至当前系统时间比如用datetime.now()生成日志路径。Airflow 的 DAG 定义再漂亮也无法保证你在本地python train.py和在 Airflow Worker 上python train.py运行的是同一份环境、同一份数据快照、同一份代码快照。Metaflow 的破局点在于它把“代码、数据、环境、元数据”四者强绑定为一个不可分割的单元这个单元叫Run。每次你执行python my_flow.py runMetaflow 并不直接调用你的函数而是捕获当前代码快照自动 diff 当前目录下所有.py文件生成 git commit hash若在 repo 中或内容哈希若不在 repo并记录冻结运行时环境通过conda-pack或pip freeze生成精确的依赖清单并打包为可复现的环境镜像本地用 conda env云端用 Docker版本化数据输入/输出每个step函数的self.input和self.output属性背后是 Metaflow 内置的DataStore默认是本地文件系统或 S3它会为每次 run 自动生成唯一 ID如run_id20240521142345789012并将该 step 的所有输出对象DataFrame、模型 pickle、JSON 字典序列化后存入s3://my-bucket/metaflow/myflow/20240521142345789012/start/output.pkl这样的路径记录完整元数据包括 start/end 时间戳、CPU/GPU 使用率、内存峰值、stdout/stderr 日志、甚至你用resources(cpu4, memory16000)声明的资源请求。提示这不是“额外功能”而是 Metaflow 的默认行为。你不需要写任何log_to_s3()或save_env_to_yaml()只要用了step和self.output这一切就自动发生。它的设计信条是“让正确的事成为最容易做的事”。2.2 为什么选择 Python 原生语法而不是 YAML/DSL很多新手第一反应是“YAML 多清晰为什么 Metaflow 要用 Python 写 workflow” 这是个好问题。答案藏在数据科学的工作习惯里。探索式开发Exploratory Development数据科学家写代码从来不是“先画流程图再填代码”。而是打开 Jupyterdf pd.read_csv(data.csv)→df.head()→df[feature].hist()→model.fit(df)→ 发现效果差 → 回头改read_csv参数加dtype{id: str}→ 再跑。这个过程是线性的、迭代的、高度依赖上下文的。YAML 无法表达这种“临时加一行 debug print”或“把中间结果存成 parquet 供下游分析”的需求。动态分支Dynamic Fan-out真实场景中分支逻辑常由数据决定。比如“如果训练集样本数 100 万则启用分布式训练否则用单机 XGBoost”。Airflow 的BranchPythonOperator写起来像在解谜而 Metaflow 中你只需step def split_data(self): if len(self.train_df) 1_000_000: self.next(self.distributed_train) else: self.next(self.single_train)这段 Python 代码在本地调试时完全可执行在云端运行时Metaflow 的 runtime 会自动解析self.next()的目标生成对应的 Step Functions 状态机。类型安全与 IDE 支持用 Python 写意味着你能享受 PyCharm 的自动补全、mypy 的类型检查、pytest 的单元测试。你可以给self.model加类型注解self.model: sklearn.ensemble.RandomForestClassifierIDE 就能提示.predict()方法。而 YAML你只能祈祷自己没拼错task_id。2.3 为什么默认集成 AWS却不锁定云厂商Metaflow 的架构是分层的Core Layer核心层是纯 Python定义了FlowSpec,Step,DataStore等抽象Runtime Layer运行时层则负责将这些抽象映射到具体基础设施。它提供了官方插件metaflow.plugins.aws对接 S3DataStore、Step Functions编排、Batch计算、Lambda轻量任务metaflow.plugins.kubernetes对接 K8s Job、S3/GCSDataStoremetaflow.plugins.local纯本地文件系统 进程用于开发调试。关键在于你的 Flow 代码my_flow.py完全不感知底层是 AWS 还是 K8s。你只需要在运行时指定--with batch或--with kubernetesMetaflow 就会自动加载对应插件完成环境打包、任务提交、状态轮询。这意味着你在本地用python my_flow.py run调试和在生产用python my_flow.py run --with batch提交代码 0 修改公司从 AWS 迁移到自建 K8s 集群只需更新 Metaflow 配置文件~/.metaflowconfig/config.json把DATASTORE_SYSROOT_S3指向新的 S3 endpoint把DEFAULT_METADATA_SERVICE_URL指向新的 Metadata ServiceFlow 代码一行不动你想在 CI/CD 中跑单元测试python my_flow.py test会自动使用 local runtime100% 隔离。注意这不是“抽象得不够深”而是刻意为之。Metaflow 认为数据科学家不该花时间学 CloudFormation 模板或 K8s YAML而应该专注在transform_features()函数的业务逻辑上。基础设施的复杂性由框架的插件层消化。3. 实操全流程拆解从“Hello World”到生产级特征工程 Pipeline3.1 环境准备三分钟搞定本地开发环境Metaflow 对环境要求极简但有两点必须明确Python 版本官方支持 3.8–3.11。我强烈建议用 3.9 或 3.10因为这两个版本在 conda-forge 上的pyarrow、pandas、scikit-learn包最稳定避免ImportError: libarrow.so.1500: cannot open shared object file这类底层链接错误。包管理Metaflow 默认用conda管理环境因其能精确控制 C 库依赖但完全兼容pip。如果你公司禁用 conda用pip install metaflow即可只是后续打包环境时需手动pip freeze requirements.txt。实操步骤Mac/Linux# 1. 创建独立 conda 环境推荐避免污染全局 conda create -n metaflow-tutorial python3.9 conda activate metaflow-tutorial # 2. 安装 Metaflow带 AWS 支持 pip install metaflow[aws] # 3. 初始化本地 Metaflow 配置关键 # 这会创建 ~/.metaflowconfig/config.json metaflow configure local # 4. 验证安装 python -c from metaflow import Flow; print(✅ Metaflow installed)实测心得metaflow configure local这一步不能跳。它会在~/.metaflowconfig/config.json中写入{ METAFLOW_DEFAULT_DATASTORE: local, METAFLOW_DEFAULT_METADATA: local, METAFLOW_DATASTORE_LOCAL_DIR: /Users/yourname/metaflow-data }这个local_dir就是你的所有 runs 数据、日志、中间产物的落盘位置。建议把它指向一个有足够空间的磁盘比如/Volumes/SSD/metaflow-data因为一个中等规模的模型训练 run中间数据可能达 10GB。3.2 第一个 FlowHello World with Data Versioning我们写一个最简 Flow但它已具备 Metaflow 的全部 DNA代码版本、数据版本、环境版本、可复现性。# hello_world.py from metaflow import FlowSpec, step, Parameter class HelloWorldFlow(FlowSpec): # 定义一个可配置参数用于演示“不同输入产生不同 run” message Parameter(message, defaultHello from Metaflow!) step def start(self): # self.input 是空的start step 无输入 # self.output 是一个 dict-like 对象用于存储本 step 输出 self.greeting f{self.message} (run_id: {self.run_id}) self.output {greeting: self.greeting} self.next(self.end) step def end(self): # 读取上一个 step 的输出 prev_output self.input print(f✅ Final output: {prev_output[greeting]}) # 也可以直接访问属性 print(f✅ Run ID: {self.run_id}) if __name__ __main__: HelloWorldFlow()运行与验证# 1. 本地运行使用 local datastore metadata python hello_world.py run # 2. 查看所有 runs python hello_world.py list # 3. 查看最新 run 的详细信息含时间、状态、参数 python hello_world.py status # 4. 查看该 run 的某个 step 的 stdout即 print 输出 python hello_world.py logs --run-id 20240521142345789012 --step start关键观察点self.run_id是自动生成的格式为YYYYMMDDhhmmssuuuuuu微秒级精度确保全球唯一self.input在startstep 是None在endstep 自动变成startstep 的self.output所有输出都存到了~/.metaflowconfig/META_DATASTORE_LOCAL_DIR/hello_world/20240521142345789012/下包括start/output.pkl序列化的 dict、start/logstdout、start/metadata.json时间戳、资源使用等如果你修改message参数再跑一次python hello_world.py run --message Hi there!会生成一个全新的 run_id且两个 run 的数据完全隔离互不影响。注意self.run_id不是字符串而是RunId对象但可以当字符串用。它的存在让你能用python hello_world.py run --origin-run-id 20240521142345789012启动一个“基于某次 run 的衍生实验”这是 A/B 测试的基石。3.3 进阶实战端到端信用卡欺诈检测 Pipeline现在我们构建一个真实场景从原始交易数据清洗到特征工程再到模型训练与评估。这个 Pipeline 将展示 Metaflow 的核心能力跨 step 数据传递、大型数据处理、外部依赖管理、资源声明。# fraud_detection.py import pandas as pd import numpy as np from sklearn.ensemble import RandomForestClassifier from sklearn.metrics import classification_report from metaflow import FlowSpec, step, Parameter, current, resources, batch, retry class FraudDetectionFlow(FlowSpec): # 可配置参数便于 A/B 测试 data_path Parameter(data_path, helpPath to raw CSV (e.g., s3://my-bucket/data/transactions.csv), defaultdata/transactions_sample.csv) n_estimators Parameter(n_estimators, helpNumber of trees in RF, default100) step def start(self): # 1. 加载原始数据支持本地或 S3 # Metaflow 内置的 S3 client 会自动处理 s3:// 路径 if self.data_path.startswith(s3://): from metaflow import S3 with S3() as s3: obj s3.get(self.data_path) # obj.blob 是 bytes用 pandas 读取 self.raw_df pd.read_csv(obj.blob) else: self.raw_df pd.read_csv(self.data_path) # 2. 数据质量检查简单示例 assert not self.raw_df.empty, Raw data is empty! assert is_fraud in self.raw_df.columns, Missing target column is_fraud # 3. 存储原始数据快照版本化 self.output {raw_df: self.raw_df} self.next(self.clean_data) step def clean_data(self): # 读取上一步的 raw_df df self.input[raw_df] # 简单清洗去除缺失值过多的列填充数值型缺失值 missing_pct df.isnull().mean() cols_to_drop missing_pct[missing_pct 0.5].index.tolist() df_clean df.drop(columnscols_to_drop) # 填充数值列 num_cols df_clean.select_dtypes(include[np.number]).columns df_clean[num_cols] df_clean[num_cols].fillna(df_clean[num_cols].median()) # 保存清洗后数据 self.cleaned_df df_clean self.output {cleaned_df: df_clean} self.next(self.feature_engineering) step resources(memory8000, cpu2) # 声明此 step 需要 8GB 内存2 CPU def feature_engineering(self): df self.input[cleaned_df] # 构造特征交易金额的对数、是否周末、每小时交易频次等 df[log_amount] np.log1p(df[amount]) df[is_weekend] (pd.to_datetime(df[transaction_time]).dt.dayofweek 5).astype(int) # 按用户 ID 统计历史交易次数作为特征 user_stats df.groupby(user_id).agg({ amount: [mean, std, count], is_fraud: mean # 历史欺诈率 }).round(3) user_stats.columns [user_mean_amount, user_std_amount, user_tx_count, user_fraud_rate] df df.merge(user_stats, left_onuser_id, right_indexTrue, howleft) # 保存特征矩阵 self.X df.drop(columns[is_fraud, transaction_id, user_id, transaction_time]) self.y df[is_fraud] self.output {X: self.X, y: self.y} self.next(self.train_model) step batch(gpu1, memory16000) # 此 step 需要 GPU16GB 内存 def train_model(self): from sklearn.model_selection import train_test_split X, y self.input[X], self.input[y] # 分割数据 X_train, X_test, y_train, y_test train_test_split( X, y, test_size0.2, random_state42, stratifyy ) # 训练模型 model RandomForestClassifier( n_estimatorsself.n_estimators, random_state42, n_jobs-1 # 利用所有 CPU ) model.fit(X_train, y_train) # 评估 y_pred model.predict(X_test) report classification_report(y_test, y_pred, output_dictTrue) # 保存模型和评估结果 self.model model self.report report self.output { model: model, report: report, test_accuracy: report[accuracy] } self.next(self.end) step def end(self): # 打印最终结果 print(f Model Accuracy: {self.report[accuracy]:.4f}) print(f Fraud Recall: {self.report[1][recall]:.4f}) print(f Run ID: {self.run_id}) print(f Parameters: n_estimators{self.n_estimators}) if __name__ __main__: FraudDetectionFlow()运行方式与说明# 方式1本地小数据集快速验证无需 AWS python fraud_detection.py run --data_path data/transactions_sample.csv # 方式2使用 AWS Batch 运行需提前配置 AWS 凭据 # 先确保 ~/.aws/credentials 已配置且有 s3:GetObject, batch:SubmitJob 权限 python fraud_detection.py run \ --data_path s3://my-bucket/data/transactions_full.csv \ --n_estimators 200 \ --with batch # 方式3查看某次 run 的所有 step 输出结构化 JSON python fraud_detection.py show --run-id 20240521142345789012 # 方式4下载模型到本地用于后续部署 python fraud_detection.py get-output --run-id 20240521142345789012 \ --step train_model \ --object model \ --path ./downloaded_model.pkl参数详解与原理装饰器/参数作用为什么重要resources(memory8000, cpu2)告诉 Metaflow 此 step 至少需要 8GB 内存、2 个 CPU 核心本地运行时Metaflow 会用psutil监控资源AWS Batch 运行时会自动映射为vcpus2, memory8192。避免 OOM Kill。batch(gpu1, memory16000)声明此 step 需要 GPU 加速Metaflow 会自动选择p3.2xlarge或g4dn.xlarge实例并挂载 NVIDIA 驱动。你不用写任何 CUDA 初始化代码。--with batch运行时切换到 AWS Batch runtime代码 0 修改只需加一个 flag。Metaflow 会自动1) 打包当前 conda env 为 Docker image2) 上传到 ECR3) 提交 Batch Job4) 轮询状态5) 下载结果到 S3。self.run_id每次 run 的唯一标识符它是所有数据、日志、元数据的 root key。你可以用它做 lineage trackingpython fraud_detection.py run --origin-run-id old_run_id启动对比实验。实测心得batch和resources是 Metaflow 最被低估的特性。很多团队花大量时间写 Kubernetes YAML 或 Terraform 模板来申请资源而 Metaflow 把它压缩成一行 Python 装饰器。更重要的是资源声明是代码的一部分会被版本化。你永远知道“20240521142345789012 这次 run 是在 p3.2xlarge 上跑的”而不是靠运维日志去猜。3.4 生产就绪如何将 Flow 部署为 API 服务Metaflow 本身不提供 HTTP server但它与主流部署方案无缝集成。最常用的是Metaflow FastAPI Docker组合实现“模型即服务”。步骤1创建 FastAPI 服务api_server.pyfrom fastapi import FastAPI, HTTPException from pydantic import BaseModel import joblib from metaflow import Flow, Run import pandas as pd app FastAPI(titleFraud Detection API) class PredictionRequest(BaseModel): amount: float user_id: str transaction_time: str # ISO format app.on_event(startup) async def load_model(): # 启动时加载最新成功的模型 flow Flow(FraudDetectionFlow) # 获取最近一次成功完成的 run latest_run None for run in flow: if run.successful and run.finished: latest_run run break if not latest_run: raise RuntimeError(No successful run found!) # 从 train_model step 加载模型 train_step latest_run[train_model] model_artifact train_step[model] app.state.model model_artifact.data app.state.run_id latest_run.id print(f✅ Loaded model from run {latest_run.id}) app.post(/predict) def predict(request: PredictionRequest): try: # 构造单行 DataFrame模拟 feature_engineering 的输出 df pd.DataFrame([{ amount: request.amount, user_id: request.user_id, transaction_time: request.transaction_time, # 其他特征需从数据库或缓存中实时获取 # 这里简化为固定值 log_amount: np.log1p(request.amount), is_weekend: 0, user_mean_amount: 100.0, user_std_amount: 50.0, user_tx_count: 10, user_fraud_rate: 0.01 }]) # 预测 prediction app.state.model.predict(df)[0] probability app.state.model.predict_proba(df)[0, 1] return { prediction: int(prediction), probability: float(probability), model_run_id: app.state.run_id } except Exception as e: raise HTTPException(status_code500, detailstr(e))步骤2编写 DockerfileFROM python:3.9-slim WORKDIR /app COPY requirements.txt . RUN pip install --no-cache-dir -r requirements.txt # 复制 Metaflow Flow 代码用于启动时加载 COPY fraud_detection.py . # 复制 API 代码 COPY api_server.py . # 暴露端口 EXPOSE 8000 # 启动命令 CMD [uvicorn, api_server:app, --host, 0.0.0.0:8000, --port, 8000]步骤3部署与调用# 1. 构建镜像需先配置 AWS 凭据因 Metaflow 需要访问 S3 docker build -t fraud-api . # 2. 运行容器本地测试 docker run -p 8000:8000 fraud-api # 3. 调用 API curl -X POST http://localhost:8000/predict \ -H Content-Type: application/json \ -d {amount: 250.0, user_id: U12345, transaction_time: 2024-05-21T14:30:00Z} # 返回: {prediction:0,probability:0.1234,model_run_id:20240521142345789012}关键设计点模型热加载app.on_event(startup)在服务启动时自动从 Metaflow 的 metadata service 查询最新成功 run并加载其模型。无需人工导出.pkl文件。Lineage 追溯API 返回model_run_id前端或监控系统可直接跳转到 Metaflow UI 查看该 run 的完整 lineage输入数据版本、代码 commit、训练参数、评估报告。零配置部署整个服务不依赖任何外部数据库或配置中心所有状态模型、元数据都来自 Metaflow 的 DataStore 和 Metadata Service。4. 常见问题排查与避坑指南那些文档里不会写的细节4.1 “ModuleNotFoundError” 为什么总在 Batch 上出现三个必查点这是新手踩坑率最高的问题。现象本地python my_flow.py run成功但python my_flow.py run --with batch失败日志里全是ModuleNotFoundError。排查顺序按优先级检查 conda 环境是否真的被打包Metaflow 默认用conda-pack打包环境。但如果你的 conda env 是用pip install安装的包而非conda installconda-pack可能漏掉某些.so文件。✅ 解决方案运行conda-pack -n your_env_name --force手动打包检查输出 tarball 是否包含lib/python3.9/site-packages/your_package/。如果缺失改用pip install --target /path/to/env/lib/python3.9/site-packages your_package。确认 AWS Batch Job Role 有 S3 权限Batch Job Role 必须有s3:GetObject权限且Resource必须包含你的 Metaflow DataStore bucket如arn:aws:s3:::my-metaflow-bucket/*。❌ 常见错误只给了s3:ListBucket忘了s3:GetObject或 Resource 写成了arn:aws:s3:::my-metaflow-bucket少了/*。检查 Python 路径是否被污染Batch 容器启动时会执行/opt/conda/bin/activate your_env。但如果your_env的site-packages里有setuptools或pip的旧版本可能覆盖 Metaflow 的依赖。✅ 解决方案在 Flow 开头加一段诊断代码step def start(self): import sys print(Python path:, sys.path) import metaflow print(Metaflow version:, metaflow.__version__) # 如果这里报错说明环境根本没加载成功4.2 如何调试“卡在某个 step 不动”Metaflow 的日志分层机制Metaflow 的日志不是扁平的而是分三层必须按顺序查日志层级查看命令适用场景Step stdout/stderrpython my_flow.py logs --run-id id --step step_name查看print()、logging.info()输出最常见的 debug 入口Batch Job 日志CloudWatchaws logs tail /aws/batch/job --follow当 step 卡住但 stdout 没输出时查 CloudWatch。常见原因ECR 镜像拉取超时、IAM Role 权限不足、S3 endpoint 配置错误Metadata Service 日志kubectl logs -n metaflow metaflow-metadata-service-xxxK8s极少数情况如 Metadata Service 本身崩溃导致 run 状态无法更新实操技巧用--quiet参数减少干扰日志。python my_flow.py logs --run-id 20240521142345789012 --step train_model --quiet只显示该 step 的原始输出不带 Metaflow 的装饰日志。4.3 “DataStore 写入失败”怎么办S3 权限与路径的隐式规则当你用--datastore s3时Metaflow 会尝试写入s3://bucket/prefix/myflow/...。但经常失败原因如下Bucket Policy 阻止了s3:PutObjectAclMetaflow 默认用ACLpublic-read上传对象为了兼容旧版 S3 客户端。如果你的 bucket policy 显式拒绝了s3:PutObjectAcl就会失败。✅ 解决方案在~/.metaflowconfig/config.json中添加S3_SERVER_SIDE_ENCRYPTION: AES256, S3_DISABLE_MPU: true, S3_PUT_OBJECT_ACL: private // 改为 private路径中包含非法字符Metaflow 会把Parameter的值直接拼进 S3 路径。如果你的--data_path是s3://my-bucket/data/2024-05-21/那么2024-05-21会被当作路径一部分。但 S3 路径中/是合法的-也是合法的真正非法的是?,#,*等。✅ 解决方案对 Parameter 值做 URL encode或在 Flow 中用urllib.parse.quote()处理。4.4 性能瓶颈为什么我的batchstep 启动要 5 分钟AWS Batch Job 启动慢90% 的原因是Docker 镜像太大。Metaflow 默认打包整个 conda env一个含tensorflow,pytorch,pandas的 env 可达 2GB拉取时间 3 分钟。优化方案三选一精简 conda env创建专用 env只装必要包conda create -n mf-prod python3.9 conda activate mf-prod pip install metaflow scikit-learn pandas numpy # ❌ 不要装 jupyter, matplotlib, seaborn训练 pipeline 不需要使用预构建的基础镜像Metaflow 官方提供metaflow/ml-base镜像含 CUDA、PyTorch、TensorFlow你只需FROM metaflow/ml-base:3.9再COPY你的代码。镜像大小从 2GB 降到 800MB。启用 ECR 层级缓存在~/.metaflowconfig/config.json中开启ECR_REGISTRY: 123456789012.dkr.ecr.us-east-1.amazonaws.com, ECR_CACHE_LAYER: true这样只有你代码的变更层会被上传基础环境层复用。4.5 安全红线如何避免在 Flow 中硬编码敏感信息Metaflow 提供了安全的 Secret 管理机制绝对不要在代码里写os.environ[AWS_SECRET_ACCESS_KEY]或password my_pass。正确做法使用secret装饰器Metaflow 2.7from metaflow import FlowSpec, step, secret class SecureFlow(FlowSpec): step def start(self): # 从 AWS Secrets Manager 加载 secret(nameprod-db-creds, typeaws-secrets-manager) def load_creds(): import json return json.loads(self.secrets[prod-db-creds]) creds load_creds() # 现在 creds 是一个 dict含 username/password self.db_url fpostgresql://{creds[username]}:{creds[password]}... self.next(self.end)前提条件在 AWS Secrets Manager 中创建 secretprod-db-creds值为{username:user,password:pass}Batch Job Role