如何用Snakemake生成机器可读报告，实现工作流执行信息的查询与入库

核心查询问题的解决思路 #

• 问题1：my_workflow1曾使用过哪些执行参数？
  在工作流执行时将参数写入专属追踪文件或日志，后续从存储的执行元数据中筛选对应工作流的参数历史记录。
• 问题2：my_workflow2以参数(1,2)执行时，任务function_two中的y值是什么？
  在任务执行过程中捕获中间变量y，将其与工作流名称、参数组合绑定存储，通过参数和任务名的组合匹配查询对应值。
• 问题3：my_workflow以参数(1,2)执行时，file_name输入任务的文件内容是什么？
  在任务执行前后，将文件内容（或内容哈希+文件路径）关联到当前工作流实例的参数组合，后续通过索引匹配查询对应记录。
• 问题4：my_workflow输出为1时，function_two的版本号/Conda环境/Git提交记录是什么？
  执行任务时自动捕获环境信息（Conda环境导出内容、Git提交哈希、任务脚本版本），并与工作流输出结果关联存储，通过输出值匹配查询对应环境数据。
• 问题5：哪些工作流执行特定任务时用了Docker容器"xyz"或scipy<1.15？
  记录每个任务的执行环境配置（Docker镜像标签、依赖包版本），通过环境特征关键词筛选关联的工作流和任务。
• 问题6：my_workflow以参数(1,2)执行时耗时多久？最耗时步骤是什么？
  利用Snakemake内置的计时功能，记录每个任务的开始/结束时间，计算总耗时和单任务耗时，排序后得到最耗时步骤。
• 问题7：最耗时步骤所在节点的架构是什么？
  在任务执行时调用系统命令（如nproc、free -h）捕获节点硬件信息，与任务计时数据关联存储。

Snakemake生成JSON类机器可读报告的实现方法 #

Snakemake默认--report生成HTML，但可通过自定义规则、日志脚本或API实现JSON格式的机器可读输出：

1. 自定义规则收集元数据 #

编写专门的Snakemake规则，在工作流结束后合并所有执行元数据，写入JSON文件：

rule generate_json_report:
    input:
        expand("{task}_metadata.json", task=all_tasks),
        workflow_params="workflow_params.json"
    output:
        "workflow_execution_report.json"
    run:
        import json
        report_data = {}
        # 合并所有任务元数据
        for infile in input:
            with open(infile) as f:
                report_data.update(json.load(f))
        # 添加工作流级参数
        with open(input.workflow_params) as f:
            report_data["workflow_global_params"] = json.load(f)
        # 写入最终报告
        with open(output[0], "w") as f:
            json.dump(report_data, f, indent=2)

2. 实时日志捕获元数据 #

通过自定义日志处理脚本，在工作流执行过程中实时收集任务的参数、环境、计时等信息，写入JSON格式追踪文件：

# log_handler.py
import json
import os
from snakemake.logging import LoggingConfig

def handle_log(record):
    task_metadata = {
        "workflow_name": os.environ.get("SNAKEMAKE_WORKFLOW_NAME"),
        "task_name": record.taskname,
        "params": record.params,
        "start_time": record.start_time.isoformat(),
        "duration": (record.end_time - record.start_time).total_seconds(),
        "conda_env": os.environ.get("CONDA_DEFAULT_ENV"),
        "git_commit": os.popen("git rev-parse HEAD").read().strip(),
        "node_cores": os.popen("nproc").read().strip(),
        "node_memory": os.popen("free -h | grep Mem | awk '{print $2}'").read().strip()
    }
    # 追加到追踪文件
    with open("workflow_tracking.json", "a") as f:
        json.dump(task_metadata, f)
        f.write("\n")

LoggingConfig.log_handler = handle_log

执行工作流时指定该脚本：

snakemake --log-handler-script log_handler.py

3. 利用Snakemake API生成结构化报告 #

通过Python脚本调用Snakemake API，在工作流执行完成后提取所有数据，生成带Schema的JSON报告：

import snakemake
import json

# 初始化工作流
workflow = snakemake.Workflow(
    snakefile="Snakefile",
    configfile="config.yaml"
)

# 执行工作流
workflow.execute()

# 构建结构化报告
report_data = {
    "schema_version": "1.0",
    "workflows": []
}
for wf in workflow.workflows:
    wf_entry = {
        "name": wf.name,
        "executions": []
    }
    for exec_instance in wf.executions:
        exec_entry = {
            "params": exec_instance.params,
            "total_duration": exec_instance.duration,
            "tasks": []
        }
        for task in exec_instance.tasks:
            task_entry = {
                "name": task.name,
                "duration": task.duration,
                "conda_env": task.conda_env,
                "docker_image": task.docker_image,
                "outputs": [str(o) for o in task.outputs]
            }
            exec_entry["tasks"].append(task_entry)
        wf_entry["executions"].append(exec_entry)
    report_data["workflows"].append(wf_entry)

# 写入带Schema的JSON报告
with open("structured_workflow_report.json", "w") as f:
    json.dump(report_data, f, indent=2)

关键注意事项 #

• 所有元数据需绑定唯一工作流实例ID（如参数组合哈希+执行时间戳），确保查询精准匹配；
• 大体积数据（如文件内容）建议存储哈希值+文件路径，避免JSON报告过大；
• 可定义统一JSON Schema规范报告结构，方便后续存入MongoDB或知识图谱。

内容的提问来源于stack exchange，提问作者Ferdi