1
$\begingroup$

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?

$\endgroup$
0
$\begingroup$

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.

$\endgroup$

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service, privacy policy and cookie policy

Not the answer you're looking for? Browse other questions tagged or ask your own question.