Как передать выходные данные из двух разных списков последовательному процессу в следующем потоке

Question

Допустим, у меня есть два процесса.

Channel
    .fromFilePairs("${params.dir}/{SPB_50k_exome_seq,FE_50k_exome_seq}.{bed,bim,fam}",size:3) {
        file -> file.baseName
    }
    .filter { key, files -> key in params.pops }
    .set { plink_data }

process pling_1 {
    publishDir "${params.outputDir}/filtered"

    input:
    set pop, file(pl_files) from plink_data

    output:
    file "${pop}_filtered.{bed,fam,bim}" into pling1_results

    script:
    output_file = "${pop}_filtered"
    base        = pl_files[0].baseName

     """
        plink2 \
        --bfile $pop \
        --hwe 0.00001 \
        --make-bed \
        --out ${output_file} \
     """
}
process pling_2 {
    publishDir "${params.outputDir}/filtered_vcf"

    input:
    set file(bed), file(bim), file(fam) from pling1_results.collect()
    file(fam1) from fam_for_plink2

    output:
    file("${base}.vcf.gz") into pling2_results

    script:

    base          = bed.baseName
    output_file   = "${base}"

     """
     plink2 \
     --bfile $base \
     --keep-fam ${params.fam}/50k_exome_seq_filtered_for_VEP_ID.txt \
     --recode vcf-iid bgz --out ${output_file}
     """
}

Результат процесса pling_1 - это два списка элементов,

[/work/SPB_50k_exome_seq.bed, /work/SPB_50k_exome_seq.bim,/work/SPB_50k_exome_seq.fam]
[/work/FE_50k_exome_seq.bed, /work/FE_50k_exome_seq.bim,/work/FE_50k_exome_seq.fam]

Следовательно, в ping_2 нет, я не способен обрабатывать SPB_50k_exome_seq и FE_50k_exome_seq в одном go. base = bed.baseName берет только SPB_50k_exome_seq и исключает FE_50k_exome_seq из второго списка. В этом случае, как я могу передать и SPB_50k_exome_seq, и FE_50k_exome_seq процессу pling_2?

Любая помощь или предложения приветствуются.

Спасибо

Answer 1

После множества экспериментов я нашел решение. Помогло бы тем, кто ищет решения.

Причина проблемы заключалась в том, что процесс pling_1 производит так много файлов в качестве вывода, и все они выводятся в один канал. Следовательно, вам нужно разбить и повторно сгруппировать эти файлы в кортеж формата. Для этого я использовал следующий канал, где вы можете использовать такие операторы, как combine & flatten.

pling1_results
    .collect()
   .flatten()
    .map { file -> tuple(file.baseName, file)}
    .groupTuple(by: 0)
    .map { input -> tuple(input[0], input[1][0], input[1][1], input[1][2])}
    .set { pl1 }

Затем используйте канал pl1 в качестве входных данных для следующего процесса. В дополнение к этому вы должны передать все элементы через канал в процесс как один вход. Подобно следующему,

set val(pop1),file(bed), file(bim), file(fam),file(fam1) from pl1.combine(fam_for_plink2)

С этими двумя изменениями рабочий процесс выполняется для каждого образца / ввода в кортеже, а не только для одного. Спасибо