如何在Nextflow中将多组Tuple通道合并为排序后的Tuple

我有三个通道 ch_lf、ch_hb 和 ch_stat2，其中 ch_lf 的格式示例如下：

[id1, linkedfrag.id1.chr1]
[id1, linkedfrag.id1.chr2]
...

我希望得到 ch_phase 通道，格式示例为：

[id1, [linkedfrag.id1.chr1,linkedfrag.id1.chr2,...], [hb_id1_chr1,hb_id2_chr2...], [stat2_id1_chr1,stat2_id1_chr2,...]]
[id2,...]

我当前的实现代码如下：

ch_lf.groupTuple().set {ch_lfs}
ch_hb.groupTuple().set {ch_hbs}
ch_stat2.groupTuple().set {ch_stat2s}

ch_phase = ch_lfs.concat(ch_hbs, ch_hetsnpvcfs).groupTuple()
  .map {id, files -> tuple(id, files[0], files[1], files[2], files[3])} //how to do this smarter

其中 ch_lfs 的格式为：

[id1, [linkedfrag.id1.chr1, linkedfrag.id1.chr2,...]]
[id2,...]

我有两个问题：

1. 从 ch_lf、ch_hb 和 ch_stat2 得到 ch_phase，有没有更优的实现方式？我尝试过 join，但它似乎只能连接两个通道；
2. 如何对 groupTuple 输出通道中的元素按染色体数字顺序排序？

1. 多通道关联的优化实现 #

你可以通过链式调用 join 来关联多个通道，这比用 concat 再 groupTuple 更清晰，还能避免因顺序问题导致的匹配错误。具体实现如下：

// 先对每个通道按id分组，得到每个id对应的文件列表
def ch_lfs = ch_lf.groupTuple()
def ch_hbs = ch_hb.groupTuple()
def ch_stat2s = ch_stat2.groupTuple()

// 链式join关联三个通道，按id精准匹配
ch_phase = ch_lfs.join(ch_hbs).join(ch_stat2s)
  .map { id, lf_list, hb_list, stat2_list -> 
      tuple(id, lf_list, hb_list, stat2_list) 
  }

链式 join 会依次按id匹配每个通道的数据，确保同一id的三组文件列表正确关联，逻辑直观且不易出错。

2. 按染色体数字排序 #

要对分组后的文件列表按染色体数字排序，需要从文件名中提取染色体编号，再按数字大小排序。可以根据你的Nextflow版本选择两种实现方式：

方法一：在groupTuple中直接指定排序规则（推荐，Nextflow 22.10+适用） #

Nextflow 22.10及以上版本支持在 groupTuple 中通过 sort 参数自定义排序逻辑：

// 定义提取染色体数字的闭包
def sortByChrNum = { file ->
    // 从文件名中提取chr后的数字，比如"linkedfrag.id1.chr12"提取12
    def chrMatch = file.name =~ /chr(\d+)/
    chrMatch ? Integer.parseInt(chrMatch[0][1]) : 0
}

// 分组时直接按染色体数字排序
def ch_lfs = ch_lf.groupTuple(sort: sortByChrNum)
def ch_hbs = ch_hb.groupTuple(sort: sortByChrNum)
def ch_stat2s = ch_stat2.groupTuple(sort: sortByChrNum)

// 链式join得到最终通道
ch_phase = ch_lfs.join(ch_hbs).join(ch_stat2s)

方法二：分组后通过map手动排序（兼容低版本Nextflow） #

如果你的Nextflow版本较低，不支持 groupTuple 的 sort 参数，可以在分组后用 map 手动排序：

def sortByChrNum = { fileList ->
    fileList.sort { file ->
        def chrMatch = file.name =~ /chr(\d+)/
        chrMatch ? Integer.parseInt(chrMatch[0][1]) : 0
    }
}

// 分组后对每个id的文件列表排序
def ch_lfs = ch_lf.groupTuple().map { id, list -> tuple(id, sortByChrNum(list)) }
def ch_hbs = ch_hb.groupTuple().map { id, list -> tuple(id, sortByChrNum(list)) }
def ch_stat2s = ch_stat2.groupTuple().map { id, list -> tuple(id, sortByChrNum(list)) }

ch_phase = ch_lfs.join(ch_hbs).join(ch_stat2s)

如果你的染色体包含非数字编号（比如chrX、chrY），可以调整排序逻辑，比如给这类染色体设置高于数字染色体的优先级，例如：

def sortByChr = { file ->
    def chrMatch = file.name =~ /chr(\w+)/
    if (!chrMatch) return 0
    def chr = chrMatch[0][1]
    chr.isInteger() ? Integer.parseInt(chr) : 
        chr == 'X' ? 23 : chr == 'Y' ? 24 : 25
}

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