Answers.Body: <p>Assuming you are using latest version of Nextflow which support the DSL 2 language, and assuming your python process output the file blah.csv, then the easiest way to move forward will be:</p>
  <pre><code>process process_a() {
      output:
      path(&quot;blah.csv&quot;)
  
      '''
      python.exe ./make_the_csv_file.py --path=blah.csv
      '''
  }
  
  process process_b() {
  
      input:
      val(the_field)
  
      '''
      echo ${the_field}
      '''
  
  }
  
  workflow do_a_then_b{
      process_a() \
          | splitCsv \
          | map{ it -&gt; it.TheFieldIWant } \
          | process_b
  }
  </code></pre>
  <p>You can also refer to the nextflow google group.</p>
  
Answers.ContentLicense: CC BY-SA 4.0
Answers.CreationDate: 2021-06-18T15:47:09.740Z
Answers.LastActivityDate: 2021-06-18T15:47:09.740Z
Answers.UserDisplayName: Sam
Body: <p>Process A runs a python script which generates a CSV file in some absolute file path say &quot;/mnt/x/blah.csv&quot;.</p>
  <p>Process B needs to consume the CSV file line by line, after process A has finished.</p>
  <p>The trouble is that the channel to input to B is initialized at the beginning before &quot;/mnt/x/blah.csv&quot; is created by A so it fails. This occurs even if I collect the result of process A as input to process B (because channel is initialized regardless)</p>
  <pre><code>nextflow.enable.dsl=2
  
  
  def get_the_csv_records() {
      Channel
          .fromPath(&quot;/mnt/x/blah.csv&quot;)
          .splitCsv(header:true, quote: '\&quot;')
          .map{ it.TheFieldIWant }
  }
  
  process process_a() {
      
      output:
      val(&quot;I'm done!&quot;)
  
      '''
      python.exe ./make_the_csv_file.py --path=/mnt/x/blah.csv
      '''
  }
  
  process process_b() {
  
      input:
      val(previous_step_is_done_dummy_variable)
      val(the_field)
  
      '''
      echo ${the_field}
      '''
  
  }
  
  workflow do_a_then_b{
      process_a()
      process_b(process_a.out.collect(), get_the_csv_records())
  }
  </code></pre>
  <p>This fails immediately with &quot;No such file: /mnt/x/blah.csv&quot; because the channel is initialized before process_a has even started.</p>
  <p>Same behavior occurs when using concat:</p>
  <pre><code>    process_b(process_a.out.collect().concat(get_the_csv_records()))
  </code></pre>
  <p>Question:</p>
  <ol>
  <li><p>How can I make process_b() only read the input channel from the CSV file after it is created?</p>
  </li>
  <li><p>Alternatively, can I make process_b() trigger when process_a() is finished some other way?</p>
  </li>
  <li><p>Alternatively, can I add an intermediate step which reads process_a.out.collect(), then reads /mnt/x/blah.csv and outputs it into some existing channel which process_b() takes as input?</p>
  </li>
  <li><p>Another approach - what is the way to create a channel which outputs the results of splitCsv, when the file is not known at compile time?</p>
  </li>
  </ol>
  
Tags: <nextflow><workflow-management>
Title: Nextflow: How to initialize channel from CSV file but only after CSV file created?
ThreadText: # Nextflow: How to initialize channel from CSV file but only after CSV file created?
  
  (Asked by: username_0 on 2021-06-05)
  
  Process A runs a python script which generates a CSV file in some absolute file path say "/mnt/x/blah.csv".
  
  Process B needs to consume the CSV file line by line, after process A has finished.
  
  The trouble is that the channel to input to B is initialized at the beginning before "/mnt/x/blah.csv" is created by A so it fails. This occurs even if I collect the result of process A as input to process B (because channel is initialized regardless)
  
  ```
  nextflow.enable.dsl=2
  
  def get_the_csv_records() {
      Channel
          .fromPath("/mnt/x/blah.csv")
          .splitCsv(header:true, quote: '\"')
          .map{ it.TheFieldIWant }
  }
  
  process process_a() {
      
      output:
      val("I'm done!")
  
      '''
      python.exe ./make_the_csv_file.py --path=/mnt/x/blah.csv
      '''
  }
  
  process process_b() {
  
      input:
      val(previous_step_is_done_dummy_variable)
      val(the_field)
  
      '''
      echo ${the_field}
      '''
  
  }
  
  workflow do_a_then_b{
      process_a()
      process_b(process_a.out.collect(), get_the_csv_records())
  }
  
  ```
  
  This fails immediately with "No such file: /mnt/x/blah.csv" because the channel is initialized before process\_a has even started.
  
  Same behavior occurs when using concat:
  
  ```
      process_b(process_a.out.collect().concat(get_the_csv_records()))
  
  ```
  
  Question:
  
  1. How can I make process\_b() only read the input channel from the CSV file after it is created?
  2. Alternatively, can I make process\_b() trigger when process\_a() is finished some other way?
  3. Alternatively, can I add an intermediate step which reads process\_a.out.collect(), then reads /mnt/x/blah.csv and outputs it into some existing channel which process\_b() takes as input?
  4. Another approach - what is the way to create a channel which outputs the results of splitCsv, when the file is not known at compile time?
  
  **Tags:** nextflow, workflow-management
  
  
  ---
  
  ## 1 answers
  
  ### Answer by username_1 on 2021-06-18. Score: 1
  
  Assuming you are using latest version of Nextflow which support the DSL 2 language, and assuming your python process output the file blah.csv, then the easiest way to move forward will be:
  
  ```
  process process_a() {
      output:
      path("blah.csv")
  
      '''
      python.exe ./make_the_csv_file.py --path=blah.csv
      '''
  }
  
  process process_b() {
  
      input:
      val(the_field)
  
      '''
      echo ${the_field}
      '''
  
  }
  
  workflow do_a_then_b{
      process_a() \
          | splitCsv \
          | map{ it -> it.TheFieldIWant } \
          | process_b
  }
  
  ```
  
  You can also refer to the nextflow google group.
  
  
  ---
  
  