如何在Snakemake中动态传递CSV至Pandas合并并解决相关技术问题？

1. 处理缺失或损坏文件的最佳实践 #

• 依赖前置校验：Snakemake核心是依赖追踪，若上游规则未生成某个data.csv，当前规则不会执行——因为Snakemake会先检查所有输入是否存在且最新。若上游规则可能失败导致文件缺失，要给上游规则添加错误处理（比如onerror回调），或用checkpoint处理动态文件列表，避免硬编码SAMPLES导致输入不存在。
• 损坏文件校验：读取CSV时加入校验逻辑，跳过无效文件：

  dfs = []
  for f in input:
      try:
          df = pd.read_csv(f)
          # 可选：校验必填列是否存在
          if {"col1", "col2"}.issubset(df.columns):
              dfs.append(df)
          else:
              print(f"跳过文件 {f}：列不匹配要求")
      except Exception as e:
          print(f"读取文件 {f} 失败：{str(e)}，跳过该文件")
  if not dfs:
      raise ValueError("无有效CSV文件可合并")
  

• 记录无效样本：可生成invalid_samples.txt作为辅助输出，记录缺失/损坏的文件路径，方便后续排查。

2. 更符合Snakemake风格的动态输入处理 #

• 自动获取样本列表：用glob_wildcards替代硬编码SAMPLES，自动从现有文件中提取样本名，避免手动维护：

  SAMPLES, = glob_wildcards("results/{sample}/data.csv")
  
  rule combine_tables:
      input:
          expand("results/{sample}/data.csv", sample=SAMPLES)
      output:
          "results/combined/all_data.csv"
      run:
          # 读取逻辑不变
  

• 用Checkpoint处理完全动态输入：若上游规则生成的样本数量不确定，用checkpoint让Snakemake先执行上游，再收集实际存在的文件：

  import os
  
  checkpoint generate_samples:
      output:
          directory("results")
      run:
          # 动态生成样本文件的逻辑
  
  def get_sample_files(wildcards):
      checkpoint_output = checkpoints.generate_samples.get_output()
      sample_names = glob_wildcards(os.path.join(checkpoint_output, "{sample}/data.csv")).sample
      return expand("results/{sample}/data.csv", sample=sample_names)
  
  rule combine_tables:
      input:
          get_sample_files
      output:
          "results/combined/all_data.csv"
      run:
          # 读取逻辑不变
  

• 通过路径提取元数据：直接从输入文件路径中提取样本名，无需额外维护样本列表，更贴合Snakemake的依赖追踪逻辑。

3. 处理CSV结构不一致的情况 #

• 统一添加样本元数据：从文件路径提取样本ID并加入DataFrame，即使结构不同也能追踪数据来源：

  import os
  import pandas as pd
  
  dfs = []
  for f in input:
      sample_id = os.path.basename(os.path.dirname(f))
      df = pd.read_csv(f)
      df["sample_id"] = sample_id
      dfs.append(df)
  combined = pd.concat(dfs, ignore_index=True, sort=False)
  

• 控制合并规则：用pd.concat的join参数选择列合并方式（inner取交集，outer取并集），用keys标记数据源：

  import os
  import pandas as pd
  
  sample_list = [os.path.basename(os.path.dirname(f)) for f in input]
  dfs = [pd.read_csv(f) for f in input]
  combined = pd.concat(dfs, keys=sample_list, names=["sample_id", "original_index"])
  # 重置索引展开样本ID
  combined = combined.reset_index(level="sample_id").reset_index(drop=True)
  

• 强制统一列结构：定义标准列列表，对齐每个DataFrame的列，填充缺失值、丢弃多余列：

  import os
  import pandas as pd
  import numpy as np
  
  STANDARD_COLUMNS = ["col1", "col2", "col3"]
  dfs = []
  for f in input:
      df = pd.read_csv(f)
      df = df.reindex(columns=STANDARD_COLUMNS, fill_value=np.nan)
      df["sample_id"] = os.path.basename(os.path.dirname(f))
      dfs.append(df)
  combined = pd.concat(dfs, ignore_index=True)
  

• 确保合并顺序：若需要按特定顺序合并，对输入文件排序后再读取：

  sorted_input = sorted(input, key=lambda x: os.path.basename(os.path.dirname(x)))
  dfs = [pd.read_csv(f) for f in sorted_input]
  

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