如何使用Nextflow dsl2处理来自S3的多个输入（yaml/json）。

huangapple
go评论114阅读模式
英文:

How to process multiple input (yaml/json) from S3 using Nextflow dsl2

问题

I see you have a complex Nextflow workflow, and you want to make several modifications. However, it's a lengthy code with multiple sections and dependencies. It's challenging to address all these modifications in a single response. It's best to tackle one specific issue or modification at a time.

If you have a specific question or need help with a particular aspect of your code or workflow, please provide more specific details, and I'll do my best to assist you.

英文:

I need to process over 1k samples with the nextflow (dsl2) pipeline in aws batch. current version of the workflow process single input per run. I'm looking workflow syntax (map tuple to iterate) to process multiple inputs to run in parralel. The inputs should be in json or yaml format, path to the input files are unique to each sample.

To preserve the input file path "s3://..." I used .fromPath in channel.

Following is my single sample input config input.yaml (-parms-file)

  1. id: HLA1001
  2. bam: s3://HLA/data/HLA1001.bam
  3. vcf: s3://HLA/data/HLA1001.vcf.gz

Workflow to run single sample input

  1. process samtools_stats {
  2.     tag "${id}"
  3.     publishDir "${params.publishdir}/${id}/samtools", mode: "copy"
  5.     input:
  6.     path bam
  7.     
  8.     output:
  9.     path "${id}.stats"
  11.     script:
  12.     """
  13.       samtools stats ${bam} > ${id}.stats
  14.     """
  15. }
  17. process mosdepth_bam {
  18.     tag "${id}"
  19.     publishDir "${params.publishdir}/${id}/mosdepth", mode: "copy"
  21.     input:
  22.     path bam
  23.     path bam_idx
  25.     output:
  26.     path "${id}.regions.bed.gz"
  28.     script:
  29.     """
  30.     mosdepth --no-per-base --by 1000 --mapq 20 --threads 4 ${id} ${bam}
  31.     """
  32. }
  34. process mosdepth_cram {
  35.     tag "${id}"
  36.     publishDir "${params.publishdir}/${id}/mosdepth", mode: "copy"
  38.     input:
  39.     path bam
  40.     path bam_idx
  41.     path reference
  42.     path reference_idx
  44.     output:
  45.     path "${id}.regions.bed.gz"
  47.     script:
  48.     """
  49.     mosdepth --no-per-base --by 1000 --mapq 20 --threads 4 --fasta ${reference} ${id} ${bam}
  51.     """
  52. }
  54. process bcftools_stats {
  55.     tag "${id}"
  56.     publishDir "${params.publishdir}/${id}/bcftools", mode: "copy"
  58.     input:
  59.     path vcf
  60.     path vcf_idx
  62.     output:
  63.     path "*"
  65.     script:
  66.     """
  67.     bcftools stats -f PASS ${vcf} > ${id}.pass.stats
  68.     """
  69. }
  71. process multiqc {
  73.     tag "${id}"
  74.     publishDir "${params.publishdir}/${id}/multiqc", mode: "copy"
  75. 	
  76.     input:
  77.     path "*"
  79.     output:
  80.     path  "multiqc_data/*", emit: multiqc_ch
  82.     script:
  83.     """
  84.     multiqc . --data-format json --enable-npm-plugin
  85.     """
  86. }
  88. process compile_metrics {
  90.     tag "${id}"
  91.     publishDir "${params.publishdir}/${id}", mode: "copy"
  93.     input:
  94.     path multiqc
  96.     output:
  97.     path "${params.id}.metrics.json", emit: compile_metrics_out
  99.     script:
  100.     """
  101.     # parse and calculate all the metrics in the multiqc output to compile
  103.     compile_metrics.py \
  104.         --multiqc_json multiqc_data.json \
  105.         --output_json ${params.id}.metrics.json \
  106.         --biosample_id ${params.id}
  107.     """
  108. }
  111. /*
  112. ----------------------------------------------------------------------
  113. WORKFLOW
  114. ---------------------------------------------------------------------
  115. */
  117. id = params.id
  119. aln_file = file ( params.bam )
  120. aln_file_type = aln_file.getExtension()
  121. vcf_file = ( params.vcf )
  122. vcf_idx = channel.fromPath(params.vcf + ".tbi", checkIfExists: true)
  124. if (aln_file_type == "bam") {
  125.     cbam = channel.fromPath(params.bam, checkIfExists: true)
  126.     cbam_idx = channel.fromPath(params.bam + ".bai", checkIfExists: true)
  127. }
  128. else if (aln_file_type == "cram") {
  129.     cbam = channel.fromPath(params.bam, checkIfExists: true)
  130.     cbam_idx = channel.fromPath(params.bam + ".crai", checkIfExists: true)
  131. }
  133. reference = channel.fromPath(params.reference, checkIfExists: true)
  134. reference_idx = channel.fromPath(params.reference + ".fai", checkIfExists: true)
  136. // main
  137. workflow {
  138.     if (aln_file_type == "bam") {
  139.         samtools_stats( bam )
  140.         mosdepth_bam( bam, bam_idx )
  141.         bcftools_stats ( vcf, vcf_idx )
  142.         multiqc( samtools_stats.out.mix( mosdepth_bam.out ).collect() )
  143.         compile_metrics(multiqc.out)
  144.     } else if (aln_file_type == "cram") {
  145.         samtools_stats( bam )
  146.         mosdepth_cram( bam, bam_idx, reference, reference_idx )
  147.         bcftools_stats ( vcf, vcf_idx )
  148.         multiqc( samtools_stats.out.mix( mosdepth_cram.out ).collect() )
  149.         compile_metrics(multiqc.out)
  150.     }
  151. }

I want to modify the workflow to run for the following multi sample input in parellel

  1. samples:
  2. -
  3.  id: HLA1001
  4.  bam: s3://HLA/data/udf/HLA1001.bam
  5.  vcf: s3://HLA/data/udf/HLA1001.vcf.gz
  6. -
  7.  id: NHLA1002
  8.  bam: s3://HLA/data/sdd/HLA1002.bam
  9.  vcf: s3://HLA/data/sdd/HLA1002.vcf.gz
  10. -
  11.  id: NHLA1003
  12.  bam: s3://HLA/data/klm/HLA1003.bam
  13.  vcf: s3://HLA/data/klm/HLA1003.vcf.gz
  14. -
  15.  id: NHLA2000
  16.  bam: s3://HLA/data/rcb/HLA2000.bam
  17.  vcf: s3://HLA/data/rcb/HLA2000.vcf.gz

The expected final output folder structure for the multiple samples..

  1. s3://mybucket/results/HLA1001/
  2. samtools/
  3. mosdepth/
  4. bcftools/
  5. multiqc/
  6. metrics/HLA1001.metrics.json
  8. s3://mybucket/results/HLA1002/
  9. samtools/
  10. mosdepth/
  11. bcftools/
  12. multiqc/
  13. metrics/HLA1002.metrics.json

The input of bam/cram, vcf and input of multiqc and compile_metrics all must fetch the same sample in every single process.

Appreciate your help! Thanks

Follwing the method answered by @steve..

Contents of main.nf: update

  1. include { compile_metrics } from './modules/compile_metrics'
  3.     Channel
  4.         .fromList( params.samples )
  5.         .map { it.biosample_id }
  6.         .set { sample_ids }
  8.     compile_metrics ( sample_ids, multiqc.out.json_data )
  9. }

Contents of modules/compile_metrics/main.nf:

  1. process compile_metrics {
  3.     tag { sample_ids }
  5.     input:
  6.     val(sample_ids)
  7.     path "multiqc_data.json"
  9.     output:
  10.     tuple val(sample_ids), path("${sample_ids}.metrics.json"), emit: compile_metrics_out
  12.     script:
  13.     """
  14.     compile_metrics.py \
  15.         --multiqc_json multiqc_data.json \
  16.         --output_json "${sample_ids}.metrics.json" \\
  17.         --biosample_id "${sample_ids}" \\
  18.     """
  19. }

Update main.nf:

  1. include { mosdepth_datamash } from './modules/mosdepth_datamash'
  3. autosomes_non_gap_regions = file( params.autosomes_non_gap_regions )
  5. mosdepth_datamash( autosomes_non_gap_regions, mosdepth_bam.out.regions.mix( mosdepth_cram.out.regions ).collect() )

Update mosdepth_datamash:

  1. process mosdepth_datamash {
  3.     tag { sample }
  5.     input:
  6.     path autosomes_non_gap_regions
  7.     tuple val(sample), path(regions)
  9.     output:
  10.     tuple val(sample), path("${sample}.mosdepth.csv"), emit: coverage
  12.     script:
  13.     """
  14.     zcat "${sample}.regions.bed.gz" | bedtools intersect -a stdin -b ${autosomes_non_gap_regions} | gzip -9c > "${sample}.regions.autosomes_non_gap_n_bases.bed.gz"
  15.     .....
  16. }

Update main.nf: fix - use queue channel instead of collect

  1. mosdepth_datamash( autosomes_non_gap_regions, mosdepth_bam.out.regions.mix( mosdepth_cram.out.regions ) )

Process verifybamid works with file instead of channel.fromPath

  1. vbi2_ud = file( params.vbi2_ud )
  2. vbi2_bed = file( params.vbi2_bed )
  3. vbi2_mean = file( params.vbi2_mean )

How to modify the channel to handle backward (previous version of workflow single sample input format) compatible of single sample input format which lacks sample key

  1. id: HLA1001
  2. bam: s3://HLA/data/HLA1001.bam
  3. vcf: s3://HLA/data/HLA1001.vcf.gz

Content of processing the input mani.nf:

  1. Channel
  2.         .fromList( params.samples )
  3.         .branch { rec ->
  4.             def aln_file = file( rec.bam )
  6.             bam: aln_file.extension == 'bam'
  7.                 def bam_idx = file( "${rec.bam}.bai" )
  9.                 return tuple( rec.id, aln_file, bam_idx )
  11.             cram: aln_file.extension == 'cram'
  12.                 def cram_idx = file( "${rec.bam}.crai" )
  14.                 return tuple( rec.id, aln_file, cram_idx )
  15.         }
  16.         .set { aln_inputs }
  17.     Channel
  18.         .fromList( params.samples )
  19.         .map { rec ->
  20.             def vcf_file = file( rec.vcf )
  21.             def vcf_idx = file( "${rec.vcf}.tbi" )
  23.             tuple( rec.id, vcf_file, vcf_idx )
  24.         }
  25.         .set { vcf_inputs }
  27.     Channel
  28.         .fromList( params.samples )
  29.         .map { it.biosample_id }
  30.         .set { sample_ids }

Updated main.nf works well

  1. INPUT format A or B:
  2. A)
  3. biosample_id: NA12878-chr14
  4. bam: s3://sample-qc/data/NA12878-chr14.bam
  6. B)
  7. samples:
  8. -
  9.  biosample_id: NA12878-chr14
  10.  bam: s3://sample-qc/data/NA12878-chr14.bam
  11. ---------------------------------------------------------------
  12. workflow {
  13. ....
  14. ....
  15. params.samples = null
  16. Channel
  17.         .fromList( params.samples )
  18.         .ifEmpty { ['biosample_id': params.biosample_id, 'bam': params.bam] }
  19.         .branch { rec ->
  20.             def aln_file = rec.bam ? file( rec.bam ) : null
  22.             bam: rec.biosample_id && aln_file?.extension == 'bam'
  23.                 def bam_idx = file( "${rec.bam}.bai" )
  25.                 return tuple( rec.biosample_id, aln_file, bam_idx )
  27.             cram: rec.biosample_id && aln_file?.extension == 'cram'
  28.                 def cram_idx = file( "${rec.bam}.crai" )
  30.                 return tuple( rec.biosample_id, aln_file, cram_idx )
  31.         }
  32.         .set { aln_inputs }
  34. Channel
  35.     .fromList( params.samples )
  36.     .ifEmpty { ['biosample_id': params.biosample_id] }
  37.     .map { it.biosample_id }
  38.     .set { sample_ids }
  40. compile_metrics ( sample_ids, multiqc.out.json_data )
  41. ...
  42. ...
  43. }

Trying other way of not duplicate the code (the above code block .ifEmpty) in each process to parse the params.sample. eg two process here required to use params.sample

  1. INPUT format A or B:
  2. A)
  3. biosample_id: NA12878-chr14
  4. bam: s3://sample-qc/data/NA12878-chr14.bam
  6. B)
  8. samples:
  9. -
  10.  biosample_id: NA12878-chr14
  11.  bam: s3://sample-qc/data/NA12878-chr14.bam
  12. -------------------------------------------------------------
  13. params.samples = ''
  14. //    params.samples = null
  16. def get_samples_list() {
  17.     if (params.samples) {
  18.         return  params.samples
  19.     }
  20.     else {
  21.     return ['biosample_id': params.biosample_id, 'bam': params.bam]
  22.     }
  23. }
  25. workflow {
  27. //    params.samples = ''
  28.     samples = get_samples_list()
  29.     ...
  30.     ...
  31.     Channel
  32.         .fromList( samples )
  33.         .branch { rec ->
  34.             def aln_file = rec.bam ? file( rec.bam ) : null
  36.             bam: rec.biosample_id && aln_file?.extension == 'bam'
  37.                 def bam_idx = file( "${rec.bam}.bai" )
  39.                 return tuple( rec.biosample_id, aln_file, bam_idx )
  41.             cram: rec.biosample_id && aln_file?.extension == 'cram'
  42.                 def cram_idx = file( "${rec.bam}.crai" )
  44.                 return tuple( rec.biosample_id, aln_file, cram_idx )
  45.         }
  46.         .set { aln_inputs }
  48.     samtools_stats_bam( aln_inputs.bam, [] )
  49.     samtools_stats_cram( aln_inputs.cram, ref_fasta )
  51.     Channel
  52.         .fromList( params.samples )
  53.         .map { it.biosample_id }
  54.         .set { sample_ids }
  56.     compile_metrics ( sample_ids, multiqc.out.json_data )
  57. }

ERROR:

  1. Workflow execution stopped with the following message:
  2. Exit status   : null
  3. Error message : Cannot invoke method branch() on null object
  4. Error report  : Cannot invoke method branch() on null object
  5. ERROR ~ Cannot invoke method branch() on null object

Calling the samples from channel to reuse works well and much better approach.

  1. Channel
  2.         .fromList( params.samples )
  3.         .ifEmpty { ['biosample_id': params.biosample_id, 'bam': params.bam] }
  4.         .set { samples }
  6.     Channel
  7.         samples.branch { rec ->
  8.         ....
  9.         
  10.     Channel
  11.         samples.map { it.biosample_id }

How to read the input.yml as an argument --input_listusing .fromList and read as list compatible with code in the 'sample channel with minimal change?
--input_list s3://mybucket/input.yaml instead of directly reading -params-file input.yaml as a list in the channel.
eg.

  1. nextflow run main.nf \
  2.     -ansi-log false \
  3. //    -params-file input.yaml \
  4.     --input_list s3://mybucket/input.yaml
  5.     -work 's3://mybucket/work' \
  6.     --publish_dir 's3://mybucket/results' \
  7.     --ref_fasta 's3://mybucket/ref.fa'

Current code
....

  1. Channel
  2. //  .fromPath( params.input_list )
  3.     .fromList( params.samples )
  4.     .ifEmpty { ['biosample_id': params.biosample_id, 'bam': params.bam] }
  5.     .set { samples }

input.yaml

  1. samples:
  2. -
  3.  id: HLA1001
  4.  bam: s3://HLA/data/udf/HLA1001.bam
  5. -
  6.  id: NHLA1002
  7.  bam: s3://HLA/data/sdd/HLA1002.bam

答案1

得分: 2

以下是您提供的内容的翻译：

使用通道，可以处理任意数量的样本，包括一个样本。以下是一种使用模块来处理BAM和CRAM输入的方法。请注意，下面的每个进程都期望一个输入tuple，其中第一个元素是样本名称或键。为了能够在下游合并通道时提供很大的帮助，我们还应确保输出具有相同的样本名称或键。以下内容未经AWS Batch测试，但至少可以帮助您入门：

main.nf的内容：

  1. include { mosdepth as mosdepth_bam } from './modules/mosdepth'
  2. include { mosdepth as mosdepth_cram } from './modules/mosdepth'
  3. include { multiqc } from './modules/multiqc'
  4. include { samtools_stats as samtools_stats_bam } from './modules/samtools'
  5. include { samtools_stats as samtools_stats_cram } from './modules/samtools'
  6. include { mosdepth_datamash } from './modules/mosdepth_datamash'```
  8. 工作流程：
  10. ```workflow {
  11.     ref_fasta = file( params.ref_fasta )
  12.     autosomes_non_gap_regions = file( params.autosomes_non_gap_regions )
  14.     Channel
  15.         .fromList( params.samples )
  16.         .ifEmpty { ['id': params.id, 'bam': params.bam] }
  17.         .branch { rec ->
  18.             def aln_file = rec.bam ? file( rec.bam ) : null
  20.             bam: rec.id && aln_file?.extension == 'bam'
  21.                 def bam_idx = file( "${rec.bam}.bai" )
  23.                 return tuple( rec.id, aln_file, bam_idx )
  25.             cram: rec.id && aln_file?.extension == 'cram'
  26.                 def cram_idx = file( "${rec.bam}.crai" )
  28.                 return tuple( rec.id, aln_file, cram_idx )
  29.         }
  30.         .set { aln_inputs }
  32.     Channel
  33.         .fromList( params.samples )
  34.         .ifEmpty { ['id': params.id, 'vcf': params.vcf] }
  35.         .branch { rec ->
  36.             def vcf_file = rec.vcf ? file( rec.vcf ) : null
  38.             output: rec.id && vcf_file
  39.                 def vcf_idx = file( "${rec.vcf}.tbi" )
  41.                 return tuple( rec.id, vcf_file, vcf_idx )
  42.         }
  43.         .set { vcf_inputs }
  45.     mosdepth_bam( aln_inputs.bam, [] )
  46.     mosdepth_cram( aln_inputs.cram, ref_fasta )
  48.     samtools_stats_bam( aln_inputs.bam, [] )
  49.     samtools_stats_cram( aln_inputs.cram, ref_fasta )
  51.     bcftools_stats( vcf_inputs )
  53.     Channel
  54.         .empty()
  55.         .mix( mosdepth_bam.out.regions )
  56.         .mix( mosdepth_cram.out.regions )
  57.         .set { mosdepth_regions }
  59.     mosdepth_datamash( mosdepth_regions, autosomes_non_gap_regions )
  61.     Channel
  62.         .empty()
  63.         .mix( mosdepth_bam.out.dists )
  64.         .mix( mosdepth_bam.out.summary )
  65.         .mix( mosdepth_cram.out.dists )
  66.         .mix( mosdepth_cram.out.summary )
  67.         .mix( samtools_stats_bam.out )
  68.         .mix( samtools_stats_cram.out )
  69.         .mix( bcftools_stats.out )
  70.         .mix( mosdepth_datamash.out )
  71.         .map { sample, files -> files }
  72.         .collect()
  73.         .set { log_files }
  75.     multiqc( log_files )
  76. }```
  78. `modules/samtools/main.nf`的内容：
  80. ```process samtools_stats {
  82.     tag { sample }
  84.     input:
  85.     tuple val(sample), path(bam), path(bai)
  86.     path ref_fasta
  88.     output:
  89.     tuple val(sample), path("${sample}.stats")
  91.     script:
  92.     def reference = ref_fasta ? /--reference "${ref_fasta}"/ : ''
  94.     """
  95.     samtools stats \\
  96.         ${reference} \\
  97.         "${bam}" \\
  98.         > "${sample}.stats"
  99.     """
  100. }```
  102. `modules/mosdepth/main.nf`的内容：
  104. ```process mosdepth {
  106.     tag { sample }
  108.     input:
  109.     tuple val(sample), path(bam), path(bai)
  110.     path ref_fasta
  112.     output:
  113.     tuple val(sample), path("*.regions.bed.gz"), emit: regions
  114.     tuple val(sample), path("*.dist.txt"), emit: dists
  115.     tuple val(sample), path("*.summary.txt"), emit: summary
  117.     script:
  118.     def fasta = ref_fasta ? /--fasta "${ref_fasta}"/ : ''
  120.     """
  121.     mosdepth \\
  122.         --no-per-base \\
  123.         --by 1000 \\
  124.         --mapq 20 \\
  125.         --threads ${task.cpus} \\
  126.         ${fasta} \\
  127.         "${sample}" \\
  128.         "${bam}"
  129.     """
  130. }```
  132. `modules/bcftools/main.nf`的内容：
  134. ```process bcftools_stats {
  136.     tag { sample }
  138.     input:
  139.     tuple val(sample), path(vcf), path(tbi)
  141.     output:
  142.     tuple val(sample), path("${sample}.pass.stats")
  144.     """
  145.     bcftools stats \\
  146.         -f PASS \\
  147.         "${vcf}" \\
  148.         > "${sample}.pass.stats"
  149.     """
  150. }```
  152. `modules/multiqc/main.nf`的内容：
  154. ```process multiqc {
  156.     input:
  157.     path 'data/*'
  159.     output:
  160.     path "multiqc_report.html", emit: report
  161.     path "multiqc_data", emit: data
  163.     """
  164.     multiqc \\
  165.         --data-format json \\
  166.         .
  167.     """
  168. }```
  170. `modules/compile_metrics/main.nf`的内容：
  172. ```process compile_metrics {
  174.     tag { sample_id }
  176.     input:
  177.     val sample_id
  178.     path multiqc_json
  180.     output:
  181.     tuple val(sample_id), path("${sample_id}.metrics.json")
  183.     """
  184.     compile_metrics.py \\
  185.         --multiqc_json "${multiqc_json}" \\
  186.         --output_json "${sample_id}.metrics.json" \\
  187.         --biosample_id "${sample_id}"
  188.     """
  189. }```
  191. `nextflow.config`的内容：
  193. ```plugins {
  195.     id 'nf-amazon'
  196. }
  198. params {
  200.     ref_fasta = null
  201.     autosomes_non_gap_regions = null
  203.     samples = null
  205.     id = null
  206.     bam = null
  207.     vcf = null
  209.     publish_dir = './results'
  210.     publish_mode = 'copy'
  211. }
  213. process {
  215.     executor = 'awsbatch'
  216.     queue = 'test-queue'
  218.     errorStrategy = 'retry'
  219.     maxRetries = 3
  221.     withName: 'samtools_stats' {
  223.         publishDir = [
  224.             path: "${params.publish_dir}/samtools",
  225.            
  227. <details>
  228. <summary>英文:</summary>
  230. With [channels](https://www.nextflow.io/docs/latest/channel.html#channels) it is possible to process any number of samples, including just one. Here&#39;s one way that use [modules](https://www.nextflow.io/docs/latest/dsl2.html#modules) to handle both BAM and CRAM inputs. Note that each process below expects an input [`tuple`](https://www.nextflow.io/docs/latest/process.html#input-type-tuple) where the first element is a sample name or key. To greatly assist with being able to merge channels downstream, we should also ensure we output tuples with the same sample name or key. The following is untested on AWS Batch, but it should at least get you started:
  232. Contents of `main.nf`:

include { bcftools_stats } from './modules/bcftools'
include { mosdepth as mosdepth_bam } from './modules/mosdepth'
include { mosdepth as mosdepth_cram } from './modules/mosdepth'
include { multiqc } from './modules/multiqc'
include { samtools_stats as samtools_stats_bam } from './modules/samtools'
include { samtools_stats as samtools_stats_cram } from './modules/samtools'
include { mosdepth_datamash } from './modules/mosdepth_datamash'

workflow {

  1. ref_fasta = file( params.ref_fasta )
  2. autosomes_non_gap_regions = file( params.autosomes_non_gap_regions )
  4. Channel
  5.     .fromList( params.samples )
  6.     .ifEmpty { [&#39;id&#39;: params.id, &#39;bam&#39;: params.bam] }
  7.     .branch { rec -&gt;
  8.         def aln_file = rec.bam ? file( rec.bam ) : null
  10.         bam: rec.id &amp;&amp; aln_file?.extension == &#39;bam&#39;
  11.             def bam_idx = file( &quot;${rec.bam}.bai&quot; )
  13.             return tuple( rec.id, aln_file, bam_idx )
  15.         cram: rec.id &amp;&amp; aln_file?.extension == &#39;cram&#39;
  16.             def cram_idx = file( &quot;${rec.bam}.crai&quot; )
  18.             return tuple( rec.id, aln_file, cram_idx )
  19.     }
  20.     .set { aln_inputs }
  22. Channel
  23.     .fromList( params.samples )
  24.     .ifEmpty { [&#39;id&#39;: params.id, &#39;vcf&#39;: params.vcf] }
  25.     .branch { rec -&gt;
  26.         def vcf_file = rec.vcf ? file( rec.vcf ) : null
  28.         output: rec.id &amp;&amp; vcf_file
  29.             def vcf_idx = file( &quot;${rec.vcf}.tbi&quot; )
  31.             return tuple( rec.id, vcf_file, vcf_idx )
  32.     }
  33.     .set { vcf_inputs }
  35. mosdepth_bam( aln_inputs.bam, [] )
  36. mosdepth_cram( aln_inputs.cram, ref_fasta )
  38. samtools_stats_bam( aln_inputs.bam, [] )
  39. samtools_stats_cram( aln_inputs.cram, ref_fasta )
  41. bcftools_stats( vcf_inputs )
  43. Channel
  44.     .empty()
  45.     .mix( mosdepth_bam.out.regions )
  46.     .mix( mosdepth_cram.out.regions )
  47.     .set { mosdepth_regions }
  49. mosdepth_datamash( mosdepth_regions, autosomes_non_gap_regions )
  51. Channel
  52.     .empty()
  53.     .mix( mosdepth_bam.out.dists )
  54.     .mix( mosdepth_bam.out.summary )
  55.     .mix( mosdepth_cram.out.dists )
  56.     .mix( mosdepth_cram.out.summary )
  57.     .mix( samtools_stats_bam.out )
  58.     .mix( samtools_stats_cram.out )
  59.     .mix( bcftools_stats.out )
  60.     .mix( mosdepth_datamash.out )
  61.     .map { sample, files -&gt; files }
  62.     .collect()
  63.     .set { log_files }
  65. multiqc( log_files )

}

  2. Contents of `modules/samtools/main.nf`:

process samtools_stats {

  1. tag { sample }
  3. input:
  4. tuple val(sample), path(bam), path(bai)
  5. path ref_fasta
  7. output:
  8. tuple val(sample), path(&quot;${sample}.stats&quot;)
  10. script:
  11. def reference = ref_fasta ? /--reference &quot;${ref_fasta}&quot;/ : &#39;&#39;
  13. &quot;&quot;&quot;
  14. samtools stats \\
  15.     ${reference} \\
  16.     &quot;${bam}&quot; \\
  17.     &gt; &quot;${sample}.stats&quot;
  18. &quot;&quot;&quot;

}

  2. Contents of `modules/mosdepth/main.nf`:

process mosdepth {

  1. tag { sample }
  3. input:
  4. tuple val(sample), path(bam), path(bai)
  5. path ref_fasta
  7. output:
  8. tuple val(sample), path(&quot;*.regions.bed.gz&quot;), emit: regions
  9. tuple val(sample), path(&quot;*.dist.txt&quot;), emit: dists
  10. tuple val(sample), path(&quot;*.summary.txt&quot;), emit: summary
  12. script:
  13. def fasta = ref_fasta ? /--fasta &quot;${ref_fasta}&quot;/ : &#39;&#39;
  15. &quot;&quot;&quot;
  16. mosdepth \\
  17.     --no-per-base \\
  18.     --by 1000 \\
  19.     --mapq 20 \\
  20.     --threads ${task.cpus} \\
  21.     ${fasta} \\
  22.     &quot;${sample}&quot; \\
  23.     &quot;${bam}&quot;
  24. &quot;&quot;&quot;

}

  2. Contents of `modules/bcftools/main.nf`:

process bcftools_stats {

  1. tag { sample }
  3. input:
  4. tuple val(sample), path(vcf), path(tbi)
  6. output:
  7. tuple val(sample), path(&quot;${sample}.pass.stats&quot;)
  9. &quot;&quot;&quot;
  10. bcftools stats \\
  11.     -f PASS \\
  12.     &quot;${vcf}&quot; \\
  13.     &gt; &quot;${sample}.pass.stats&quot;
  14. &quot;&quot;&quot;

}

  2. Contents of `modules/multiqc/main.nf`:

process multiqc {

  1. input:
  2. path &#39;data/*&#39;
  4. output:
  5. path &quot;multiqc_report.html&quot;, emit: report
  6. path &quot;multiqc_data&quot;, emit: data
  8. &quot;&quot;&quot;
  9. multiqc \\
  10.     --data-format json \\
  11.     .
  12. &quot;&quot;&quot;

}

  2. Contents of `modules/compile_metrics/main.nf`:

process compile_metrics {

  1. tag { sample_id }
  3. input:
  4. val sample_id
  5. path multiqc_json
  7. output:
  8. tuple val(sample_id), path(&quot;${sample_id}.metrics.json&quot;)
  10. &quot;&quot;&quot;
  11. compile_metrics.py \\
  12.     --multiqc_json &quot;${multiqc_json}&quot; \\
  13.     --output_json &quot;${sample_id}.metrics.json&quot; \\
  14.     --biosample_id &quot;${sample_id}&quot;
  15. &quot;&quot;&quot;

}

  2. Contents of `./modules/mosdepth_datamash/main.nf`:

process mosdepth_datamash {

  1. tag { sample_id }
  3. input:
  4. tuple val(sample_id), path(regions_bed)
  5. path autosomes_non_gap_regions
  7. output:
  8. tuple val(sample_id), path(&quot;${sample_id}.mosdepth.csv&quot;)
  10. &quot;&quot;&quot;
  11. zcat -f &quot;${regions_bed}&quot; |
  12.     bedtools intersect -a stdin -b &quot;${autosomes_non_gap_regions}&quot; |
  13.     gzip -9 &gt; &quot;${sample_id}.regions.autosomes_non_gap_n_bases.bed.gz&quot;
  15. # do something
  17. touch &quot;${sample_id}.mosdepth.csv&quot;
  18. &quot;&quot;&quot;

}

  2. Contents of `nextflow.config`:

plugins {

  1. id &#39;nf-amazon&#39;

}

params {

  1. ref_fasta = null
  2. autosomes_non_gap_regions = null
  4. samples = null
  6. id = null
  7. bam = null
  8. vcf = null
  10. publish_dir = &#39;./results&#39;
  11. publish_mode = &#39;copy&#39;

}

process {

  1. executor = &#39;awsbatch&#39;
  2. queue = &#39;test-queue&#39;
  4. errorStrategy = &#39;retry&#39;
  5. maxRetries = 3
  7. withName: &#39;samtools_stats&#39; {
  9.     publishDir = [
  10.         path: &quot;${params.publish_dir}/samtools&quot;,
  11.         mode: params.publish_mode,
  12.     ]
  13. }
  15. withName: &#39;bcftools_stats&#39; {
  17.     publishDir = [
  18.         path: &quot;${params.publish_dir}/bcftools&quot;,
  19.         mode: params.publish_mode,
  20.     ]
  21. }
  23. withName: &#39;mosdepth&#39; {
  25.     cpus = 4
  27.     publishDir = [
  28.         path: &quot;${params.publish_dir}/mosdepth&quot;,
  29.         mode: params.publish_mode,
  30.     ]
  31. }
  33. withName: &#39;multiqc&#39; {
  35.     publishDir = [
  36.         path: &quot;${params.publish_dir}/multiqc&quot;,
  37.         mode: params.publish_mode,
  38.     ]
  39. }

}

aws {

  1. region = &#39;us-east-1&#39;
  3. batch {
  4.     cliPath = &#39;/home/ec2-user/miniconda/bin/aws&#39;
  5. }

}

  2. And run using something like:

$ nextflow run main.nf
-ansi-log false
-params-file input.yaml
-work 's3://mybucket/work'
--publish_dir 's3://mybucket/results'
--ref_fasta 's3://mybucket/ref.fa'

  2. </details>

huangapple
  • 本文由 发表于 2023年5月10日 19:16:30
  • 转载请务必保留本文链接：https://go.coder-hub.com/76217754.html
  • bash
  • groovy
  • nextflow
  • python
  • shell
匿名

发表评论

匿名网友

:?: :razz: :sad: :evil: :!: :smile: :oops: :grin: :eek: :shock: :???: :cool: :lol: :mad: :twisted: :roll: :wink: :idea: :arrow: :neutral: :cry: :mrgreen:

确定