Teil 3: Hello Workflow

KI-gestützte Übersetzung - mehr erfahren & Verbesserungen vorschlagen

Die meisten realen Workflows umfassen mehr als einen Schritt. In diesem Trainingsmodul lernst du, wie du processes in einem mehrstufigen Workflow verbindest.

Dies wird dir den Nextflow-Weg beibringen, um Folgendes zu erreichen:

1. Daten von einem process zum nächsten fließen lassen
2. Ausgaben von mehreren process-Aufrufen in einen einzelnen process-Aufruf sammeln
3. Mehr als eine Eingabe an einen process übergeben
4. Mehrere Ausgaben aus einem process handhaben

Zur Demonstration werden wir weiterhin auf dem domänenunabhängigen Hello World-Beispiel aus den Teilen 1 und 2 aufbauen. Diesmal werden wir die folgenden Änderungen an unserem Workflow vornehmen, um besser widerzuspiegeln, wie Leute tatsächliche Workflows erstellen:

1. Einen zweiten Schritt hinzufügen, der die Begrüßung in Großbuchstaben umwandelt.
2. Einen dritten Schritt hinzufügen, der alle transformierten Begrüßungen sammelt und sie in eine einzelne Datei schreibt.
3. Einen Parameter hinzufügen, um die endgültige Ausgabedatei zu benennen, und diesen als sekundäre Eingabe an den Sammelschritt übergeben.
4. Den Sammelschritt auch eine einfache Statistik über das Verarbeitete berichten lassen.

Wie du von diesem Abschnitt aus beginnen kannst

Dieser Abschnitt des Kurses setzt voraus, dass du die Teile 1-2 des Hello Nextflow-Kurses abgeschlossen hast, aber wenn du mit den dort behandelten Grundlagen vertraut bist, kannst du von hier aus starten, ohne etwas Besonderes tun zu müssen.

---

0. Aufwärmübung: hello-workflow.nf ausführen

Wir werden das Workflow-Skript hello-workflow.nf als Ausgangspunkt verwenden. Es entspricht dem Skript, das durch das Durcharbeiten von Teil 2 dieses Trainingskurses erstellt wurde, außer dass wir die view()-Anweisungen entfernt und das Ausgabeziel geändert haben:

output {
    first_output {
        path 'hello_workflow'
        mode 'copy'
    }
}

Um sicherzustellen, dass alles funktioniert, führe das Skript einmal aus, bevor du Änderungen vornimmst:

nextflow run hello-workflow.nf

Befehlsausgabe

 N E X T F L O W   ~  version 25.10.2

Launching `hello-workflow.nf` [admiring_lamarr] DSL2 - revision: 4d4053520d

executor >  local (3)
[b1/5826b5] process > sayHello (2) [100%] 3 of 3 ✔

Wie zuvor findest du die Ausgabedateien an dem im output-Block angegebenen Ort. Für dieses Kapitel ist es unter results/hello_workflow/.

Verzeichnisinhalte

results/hello_workflow
├── Bonjour-output.txt
├── Hello-output.txt
└── Holà-output.txt

Wenn das bei dir funktioniert hat, bist du bereit zu lernen, wie man einen mehrstufigen Workflow zusammenstellt.

---

1. Einen zweiten Schritt zum Workflow hinzufügen

Wir werden einen Schritt hinzufügen, um jede Begrüßung in Großbuchstaben umzuwandeln.

Dazu müssen wir drei Dinge tun:

• Den Befehl definieren, den wir für die Großbuchstaben-Umwandlung verwenden werden.
• Einen neuen process schreiben, der den Großbuchstaben-Befehl umhüllt.
• Den neuen process im workflow-Block aufrufen und so einrichten, dass er die Ausgabe des sayHello()-process als Eingabe nimmt.

1.1. Den Großbuchstaben-Befehl definieren und im Terminal testen

Für die Umwandlung der Begrüßungen in Großbuchstaben werden wir ein klassisches UNIX-Tool namens tr für 'text replacement' verwenden, mit der folgenden Syntax:

Syntax

tr '[a-z]' '[A-Z]'

Dies ist ein sehr naiver Textersetzungs-Einzeiler, der keine akzentuierten Buchstaben berücksichtigt, also wird zum Beispiel 'Holà' zu 'HOLà', aber er wird für die Demonstration der Nextflow-Konzepte gut genug funktionieren, und das ist, was zählt.

Um es auszuprobieren, können wir den Befehl echo 'Hello World' ausführen und seine Ausgabe an den tr-Befehl pipen:

echo 'Hello World' | tr '[a-z]' '[A-Z]' > UPPER-output.txt

Die Ausgabe ist eine Textdatei namens UPPER-output.txt, die die Großbuchstaben-Version des Hello World-Strings enthält.

Dateiinhalte

UPPER-output.txt

HELLO WORLD

Das ist im Grunde das, was wir mit unserem Workflow versuchen werden zu tun.

1.2. Den Großbuchstaben-Schritt als Nextflow process schreiben

Wir können unseren neuen process nach dem ersten modellieren, da wir alle dieselben Komponenten verwenden wollen.

Füge die folgende process-Definition zum Workflow-Skript hinzu, direkt unter dem ersten:

/*
 * Verwende ein Textersetzungstool, um die Begrüßung in Großbuchstaben umzuwandeln
 */
process convertToUpper {

    input:
    path input_file

    output:
    path "UPPER-${input_file}"

    script:
    """
    cat '$input_file' | tr '[a-z]' '[A-Z]' > 'UPPER-${input_file}'
    """
}

In diesem Fall setzen wir den zweiten Ausgabedateinamen basierend auf dem Eingabedateinamen zusammen, ähnlich wie wir es ursprünglich für die Ausgabe des ersten process getan haben.

1.3. Einen Aufruf des neuen process im workflow-Block hinzufügen

Jetzt müssen wir Nextflow sagen, dass es den process, den wir gerade definiert haben, tatsächlich aufrufen soll.

Nimm im workflow-Block die folgende Codeänderung vor:

NachherVorher

workflow {

    main:
    // Einen Channel für Eingaben aus einer CSV-Datei erstellen
    greeting_ch = channel.fromPath(params.input)
                        .splitCsv()
                        .map { line -> line[0] }
    // Eine Begrüßung ausgeben
    sayHello(greeting_ch)
    // Die Begrüßung in Großbuchstaben umwandeln
    convertToUpper()

    publish:
    first_output = sayHello.out
}

workflow {

    main:
    // Einen Channel für Eingaben aus einer CSV-Datei erstellen
    greeting_ch = channel.fromPath(params.input)
                        .splitCsv()
                        .map { line -> line[0] }
    // Eine Begrüßung ausgeben
    sayHello(greeting_ch)

    publish:
    first_output = sayHello.out
}

Dies ist noch nicht funktionsfähig, weil wir nicht angegeben haben, was in den convertToUpper()-process eingegeben werden soll.

1.4. Die Ausgabe des ersten process an den zweiten process übergeben

Jetzt müssen wir die Ausgabe des sayHello()-process in den convertToUpper()-process fließen lassen.

Praktischerweise verpackt Nextflow automatisch die Ausgabe eines process in einen channel namens <process>.out. Also ist die Ausgabe des sayHello-process ein channel namens sayHello.out, den wir direkt in den Aufruf von convertToUpper() einstecken können.

Nimm im workflow-Block die folgende Codeänderung vor:

NachherVorher

    // Die Begrüßung in Großbuchstaben umwandeln
    convertToUpper(sayHello.out)

    // Die Begrüßung in Großbuchstaben umwandeln
    convertToUpper()

Für einen einfachen Fall wie diesen (eine Ausgabe zu einer Eingabe) ist das alles, was wir tun müssen, um zwei processes zu verbinden!

1.5. Die Workflow-Ausgabeveröffentlichung einrichten

Zum Schluss aktualisieren wir die Workflow-Ausgaben, um auch die Ergebnisse des zweiten process zu veröffentlichen.

1.5.1. Den publish:-Abschnitt des workflow-Blocks aktualisieren

Nimm im workflow-Block die folgende Codeänderung vor:

NachherVorher

    publish:
    first_output = sayHello.out
    uppercased = convertToUpper.out
}

    publish:
    first_output = sayHello.out
}

Die Logik ist dieselbe wie zuvor.

1.5.2. Den output-Block aktualisieren

Nimm im output-Block die folgende Codeänderung vor:

NachherVorher

output {
    first_output {
        path 'hello_workflow'
        mode 'copy'
    }
    uppercased {
        path 'hello_workflow'
        mode 'copy'
    }
}

output {
    first_output {
        path 'hello_workflow'
        mode 'copy'
    }
}

Wieder ist die Logik dieselbe wie zuvor.

Dies zeigt dir, dass du die Ausgabeeinstellungen auf einer sehr granularen Ebene kontrollieren kannst, für jede einzelne Ausgabe. Versuche gerne, die Pfade oder den Veröffentlichungsmodus für einen der processes zu ändern, um zu sehen, was passiert.

Natürlich bedeutet das, dass wir hier einige Informationen wiederholen, was unpraktisch werden könnte, wenn wir den Speicherort für alle Ausgaben auf dieselbe Weise aktualisieren wollten. Später im Kurs lernst du, wie du diese Einstellungen für mehrere Ausgaben auf strukturierte Weise konfigurierst.

1.6. Den Workflow mit -resume ausführen

Lass uns das mit dem -resume-Flag testen, da wir den ersten Schritt des Workflows bereits erfolgreich ausgeführt haben.

nextflow run hello-workflow.nf -resume

Befehlsausgabe

 N E X T F L O W   ~  version 25.10.2

Launching `hello-workflow.nf` [high_cantor] DSL2 - revision: d746983511

executor >  local (3)
[ab/816321] process > sayHello (3)       [100%] 3 of 3, cached: 3 ✔
[e0/ecf81b] process > convertToUpper (3) [100%] 3 of 3 ✔

In der Konsolenausgabe gibt es jetzt eine zusätzliche Zeile, die dem neuen process entspricht, den wir gerade hinzugefügt haben.

Du findest die Ausgaben im Verzeichnis results/hello_workflow, wie im output-Block festgelegt.

Verzeichnisinhalte

results/hello_workflow/
├── Bonjour-output.txt
├── Hello-output.txt
├── Holà-output.txt
├── UPPER-Bonjour-output.txt
├── UPPER-Hello-output.txt
└── UPPER-Holà-output.txt

Das ist praktisch! Aber es lohnt sich trotzdem, einen Blick in das work-Verzeichnis eines der Aufrufe des zweiten process zu werfen.

Verzeichnisinhalte

work/e0/ecf81b4cacc648b9b994218d5b29d7/
├── Holà-output.txt -> /workspaces/training/hello-nextflow/work/ab/81632178cd37e9e815959278808819/Holà-output.txt
└── UPPER-Holà-output.txt

Beachte, dass es zwei *-output-Dateien gibt: die Ausgabe des ersten process sowie die Ausgabe des zweiten.

Die Ausgabe des ersten process ist dort, weil Nextflow sie dort gestageed hat, um alles Notwendige für die Ausführung innerhalb desselben Unterverzeichnisses zu haben.

Es handelt sich jedoch tatsächlich um einen symbolischen Link, der auf die Originaldatei im Unterverzeichnis des ersten process-Aufrufs zeigt. Standardmäßig verwendet Nextflow, wenn es auf einer einzelnen Maschine wie hier ausgeführt wird, symbolische Links anstelle von Kopien, um Eingabe- und Zwischendateien zu stagen.

Bevor du weitermachst, denke darüber nach, wie wir nur die Ausgabe von sayHello mit der Eingabe von convertToUpper verbunden haben und die beiden processes nacheinander ausgeführt werden konnten. Nextflow hat die harte Arbeit erledigt, einzelne Eingabe- und Ausgabedateien zu handhaben und sie zwischen den beiden Befehlen für uns weiterzugeben.

Das ist einer der Gründe, warum Nextflow channels so mächtig sind: Sie kümmern sich um die Routinearbeit, die beim Verbinden von Workflow-Schritten anfällt.

Zusammenfassung

Du weißt, wie du processes verketten kannst, indem du die Ausgabe eines Schritts als Eingabe für den nächsten Schritt bereitstellst.

Was kommt als Nächstes?

Lerne, wie du Ausgaben von gestapelten process-Aufrufen sammelst und sie in einen einzelnen process einspeist.

---

2. Einen dritten Schritt hinzufügen, um alle Begrüßungen zu sammeln

Wenn wir einen process verwenden, um eine Transformation auf jedes der Elemente in einem channel anzuwenden, wie wir es hier mit den mehreren Begrüßungen tun, wollen wir manchmal Elemente aus dem Ausgabe-channel dieses process sammeln und sie in einen anderen process einspeisen, der eine Art Analyse oder Zusammenfassung durchführt.

Zur Demonstration werden wir einen neuen Schritt zu unserer Pipeline hinzufügen, der alle vom convertToUpper-process produzierten Großbuchstaben-Begrüßungen sammelt und sie in eine einzelne Datei schreibt.

Um die Überraschung nicht zu verderben, aber das wird einen sehr nützlichen Operator beinhalten.

2.1. Den Sammelbefehl definieren und im Terminal testen

Der Sammelschritt, den wir zu unserem Workflow hinzufügen wollen, wird den cat-Befehl verwenden, um mehrere Großbuchstaben-Begrüßungen in eine einzelne Datei zu verketten.

Lass uns den Befehl für sich allein im Terminal ausführen, um zu überprüfen, dass er wie erwartet funktioniert, genau wie wir es zuvor getan haben.

Führe Folgendes in deinem Terminal aus:

echo 'Hello' | tr '[a-z]' '[A-Z]' > UPPER-Hello-output.txt
echo 'Bonjour' | tr '[a-z]' '[A-Z]' > UPPER-Bonjour-output.txt
echo 'Holà' | tr '[a-z]' '[A-Z]' > UPPER-Holà-output.txt
cat UPPER-Hello-output.txt UPPER-Bonjour-output.txt UPPER-Holà-output.txt > COLLECTED-output.txt

Die Ausgabe ist eine Textdatei namens COLLECTED-output.txt, die die Großbuchstaben-Versionen der ursprünglichen Begrüßungen enthält.

Dateiinhalte

COLLECTED-output.txt

HELLO
BONJOUR
HOLà

Das ist das Ergebnis, das wir mit unserem Workflow erreichen wollen.

2.2. Einen neuen process für den Sammelschritt erstellen

Lass uns einen neuen process erstellen und ihn collectGreetings() nennen. Wir können beginnen, ihn basierend auf dem zu schreiben, was wir bisher gesehen haben.

2.2.1. Die 'offensichtlichen' Teile des process schreiben

Füge die folgende process-Definition zum Workflow-Skript hinzu:

/*
 * Großbuchstaben-Begrüßungen in einer einzelnen Ausgabedatei sammeln
 */
process collectGreetings {

    input:
    ???

    output:
    path "COLLECTED-output.txt"

    script:
    """
    cat ??? > 'COLLECTED-output.txt'
    """
}

Das ist, was wir mit Zuversicht basierend auf dem, was du bisher gelernt hast, schreiben können. Aber das ist nicht funktionsfähig! Es lässt die Eingabedefinition(en) und die erste Hälfte des Skriptbefehls aus, weil wir herausfinden müssen, wie wir das schreiben.

2.2.2. Eingaben für collectGreetings() definieren

Wir müssen die Begrüßungen von allen Aufrufen des convertToUpper()-process sammeln. Was wissen wir, dass wir vom vorherigen Schritt im Workflow bekommen können?

Der von convertToUpper() ausgegebene channel wird die Pfade zu den einzelnen Dateien enthalten, die die Großbuchstaben-Begrüßungen enthalten. Das entspricht einem Eingabeslot; nennen wir ihn der Einfachheit halber input_files.

Nimm im process-Block die folgende Codeänderung vor:

NachherVorher

      input:
      path input_files

      input:
      ???

Beachte, dass wir das path-Präfix verwenden, obwohl wir erwarten, dass dies mehrere Dateien enthält.

2.2.3. Den Verkettungsbefehl zusammenstellen

Hier könnten die Dinge etwas knifflig werden, weil wir in der Lage sein müssen, eine beliebige Anzahl von Eingabedateien zu handhaben. Konkret können wir den Befehl nicht im Voraus schreiben, also müssen wir Nextflow sagen, wie es ihn zur Laufzeit basierend auf den Eingaben, die in den process fließen, zusammenstellen soll.

Mit anderen Worten, wenn wir einen Eingabe-channel haben, der das Element [file1.txt, file2.txt, file3.txt] enthält, brauchen wir Nextflow, um das in cat file1.txt file2.txt file3.txt umzuwandeln.

Glücklicherweise macht Nextflow das gerne für uns, wenn wir einfach cat ${input_files} in den Skriptbefehl schreiben.

Nimm im process-Block die folgende Codeänderung vor:

NachherVorher

    script:
    """
    cat ${input_files} > 'COLLECTED-output.txt'
    """

    script:
    """
    cat ??? > 'COLLECTED-output.txt'
    """

Theoretisch sollte dies jede beliebige Anzahl von Eingabedateien handhaben können.

Tipp

Einige Kommandozeilenwerkzeuge erfordern die Angabe eines Arguments (wie -input) für jede Eingabedatei. In diesem Fall müssten wir ein bisschen zusätzliche Arbeit leisten, um den Befehl zusammenzustellen. Du kannst ein Beispiel dafür im Nextflow for Genomics-Trainingskurs sehen.

2.3. Den Sammelschritt zum Workflow hinzufügen

Jetzt sollten wir nur noch den Sammel-process auf der Ausgabe des Großbuchstaben-Schritts aufrufen müssen.

2.3.1. Die process-Aufrufe verbinden

Nimm im workflow-Block die folgende Codeänderung vor:

NachherVorher

    // Die Begrüßung in Großbuchstaben umwandeln
    convertToUpper(sayHello.out)

    // Alle Begrüßungen in einer Datei sammeln
    collectGreetings(convertToUpper.out)
}

    // Die Begrüßung in Großbuchstaben umwandeln
    convertToUpper(sayHello.out)
}

Dies verbindet die Ausgabe von convertToUpper() mit der Eingabe von collectGreetings().

2.3.2. Den Workflow mit -resume ausführen

Lass uns es versuchen.

nextflow run hello-workflow.nf -resume

Befehlsausgabe

N E X T F L O W   ~  version 25.10.2

Launching `hello-workflow.nf` [mad_gilbert] DSL2 - revision: 6acfd5e28d

executor >  local (3)
[79/33b2f0] sayHello (2)         | 3 of 3, cached: 3 ✔
[99/79394f] convertToUpper (3)   | 3 of 3, cached: 3 ✔
[47/50fe4a] collectGreetings (1) | 3 of 3 ✔

Es läuft erfolgreich, einschließlich des dritten Schritts.

Schau dir jedoch die Anzahl der Aufrufe für collectGreetings() in der letzten Zeile an. Wir haben nur einen erwartet, aber es sind drei.

Schau dir jetzt den Inhalt der endgültigen Ausgabedatei an.

Dateiinhalte

results/COLLECTED-output.txt

Holà

Oh nein. Der Sammelschritt wurde einzeln auf jede Begrüßung ausgeführt, was NICHT das ist, was wir wollten.

Wir müssen etwas tun, um Nextflow explizit zu sagen, dass wir wollen, dass dieser dritte Schritt auf allen Elementen im vom convertToUpper() ausgegebenen channel läuft.

2.4. Einen Operator verwenden, um die Begrüßungen in eine einzelne Eingabe zu sammeln

Ja, wieder einmal ist die Antwort auf unser Problem ein Operator.

Konkret werden wir den treffend benannten collect()-Operator verwenden.

2.4.1. Den collect()-Operator hinzufügen

Diesmal wird es etwas anders aussehen, weil wir den Operator nicht im Kontext einer channel factory hinzufügen; wir fügen ihn zu einem Ausgabe-channel hinzu.

Wir nehmen das convertToUpper.out und hängen den collect()-Operator an, was uns convertToUpper.out.collect() gibt. Wir können das direkt in den collectGreetings()-process-Aufruf einstecken.

Nimm im workflow-Block die folgende Codeänderung vor:

NachherVorher

    // Alle Begrüßungen in einer Datei sammeln
    collectGreetings(convertToUpper.out.collect())
}

    // Alle Begrüßungen in einer Datei sammeln
    collectGreetings(convertToUpper.out)
}

2.4.2. Einige view()-Anweisungen hinzufügen

Lass uns auch ein paar view()-Anweisungen einfügen, um die Zustände des channel-Inhalts vor und nach zu visualisieren.

NachherVorher

    // Alle Begrüßungen in einer Datei sammeln
    collectGreetings(convertToUpper.out.collect())

    // Optionale view-Anweisungen
    convertToUpper.out.view { contents -> "Before collect: $contents" }
    convertToUpper.out.collect().view { contents -> "After collect: $contents" }
}

    // Alle Begrüßungen in einer Datei sammeln
    collectGreetings(convertToUpper.out.collect())
}

Die view()-Anweisungen können überall platziert werden, wo du möchtest; wir haben sie für die Lesbarkeit direkt nach dem Aufruf platziert.

2.4.3. Den Workflow erneut mit -resume ausführen

Lass uns es versuchen:

nextflow run hello-workflow.nf -resume

Befehlsausgabe

N E X T F L O W   ~  version 25.10.2

Launching `hello-workflow.nf` [soggy_franklin] DSL2 - revision: bc8e1b2726

[d6/cdf466] sayHello (1)       | 3 of 3, cached: 3 ✔
[99/79394f] convertToUpper (2) | 3 of 3, cached: 3 ✔
[1e/83586c] collectGreetings   | 1 of 1 ✔
Before collect: /workspaces/training/hello-nextflow/work/b3/d52708edba8b864024589285cb3445/UPPER-Bonjour-output.txt
Before collect: /workspaces/training/hello-nextflow/work/99/79394f549e3040dfc2440f69ede1fc/UPPER-Hello-output.txt
Before collect: /workspaces/training/hello-nextflow/work/aa/56bfe7cf00239dc5badc1d04b60ac4/UPPER-Holà-output.txt
After collect: [/workspaces/training/hello-nextflow/work/b3/d52708edba8b864024589285cb3445/UPPER-Bonjour-output.txt, /workspaces/training/hello-nextflow/work/99/79394f549e3040dfc2440f69ede1fc/UPPER-Hello-output.txt, /workspaces/training/hello-nextflow/work/aa/56bfe7cf00239dc5badc1d04b60ac4/UPPER-Holà-output.txt]

Es läuft erfolgreich, obwohl die Log-Ausgabe möglicherweise etwas unordentlicher aussieht als dies (wir haben es für die Lesbarkeit aufgeräumt).

Diesmal wurde der dritte Schritt nur einmal aufgerufen!

Wenn du dir die Ausgabe der view()-Anweisungen anschaust, siehst du Folgendes:

• Drei Before collect:-Anweisungen, eine für jede Begrüßung: zu diesem Zeitpunkt sind die Dateipfade einzelne Elemente im channel.
• Eine einzelne After collect:-Anweisung: die drei Dateipfade sind jetzt in ein einzelnes Element verpackt.

Schau dir den Inhalt der endgültigen Ausgabedatei an.

Dateiinhalte

results/COLLECTED-output.txt

BONJOUR
HELLO
HOLà

Diesmal haben wir alle drei Begrüßungen in der endgültigen Ausgabedatei. Erfolg!

Hinweis

Wenn du dies mehrmals ohne -resume ausführst, wirst du sehen, dass sich die Reihenfolge der Begrüßungen von einem Lauf zum nächsten ändert. Dies zeigt dir, dass die Reihenfolge, in der Elemente durch process-Aufrufe fließen, nicht garantiert konsistent ist.

2.4.4. view()-Anweisungen für die Lesbarkeit entfernen

Bevor du zum nächsten Abschnitt übergehst, empfehlen wir, die view()-Anweisungen zu löschen, um die Konsolenausgabe nicht zu überladen.

NachherVorher

    // Alle Begrüßungen in einer Datei sammeln
    collectGreetings(convertToUpper.out.collect())

    // Alle Begrüßungen in einer Datei sammeln
    collectGreetings(convertToUpper.out.collect())

    // Optionale view-Anweisungen
    convertToUpper.out.view { contents -> "Before collect: $contents" }
    convertToUpper.out.collect().view { contents -> "After collect: $contents" }

Dies ist im Grunde die umgekehrte Operation von Punkt 2.4.2.

Zusammenfassung

Du weißt, wie du Ausgaben aus einem Stapel von process-Aufrufen sammelst und sie in einen gemeinsamen Analyse- oder Zusammenfassungsschritt einspeist.

Was kommt als Nächstes?

Lerne, wie du mehr als eine Eingabe an einen process übergibst.

---

3. Mehr als eine Eingabe an einen process übergeben

Wir wollen in der Lage sein, der endgültigen Ausgabedatei einen bestimmten Namen zu geben, um nachfolgende Stapel von Begrüßungen verarbeiten zu können, ohne die endgültigen Ergebnisse zu überschreiben.

Zu diesem Zweck werden wir die folgenden Verfeinerungen am Workflow vornehmen:

• Den Sammel-process so modifizieren, dass er einen benutzerdefinierten Namen für die Ausgabedatei akzeptiert
• Einen Kommandozeilenparameter zum Workflow hinzufügen und ihn an den Sammel-process übergeben

3.1. Den Sammel-process modifizieren

Wir müssen die zusätzliche Eingabe deklarieren und sie in den Ausgabedateinamen integrieren.

3.1.1. Die zusätzliche Eingabe deklarieren

Gute Nachricht: Wir können so viele Eingabevariablen deklarieren, wie wir wollen, in der process-Definition. Nennen wir diese batch_name.

Nimm im process-Block die folgende Codeänderung vor:

NachherVorher

    input:
    path input_files
    val batch_name

    input:
    path input_files

Du kannst deine processes so einrichten, dass sie so viele Eingaben erwarten, wie du möchtest. Im Moment sind diese alle als erforderliche Eingaben eingerichtet; du musst einen Wert angeben, damit der Workflow funktioniert.

Die Handhabung erforderlicher und optionaler Eingaben behandeln fortgeschrittene Kurse.

3.1.2. Die batch_name-Variable im Ausgabedateinamen verwenden

Wir können die Variable auf dieselbe Weise in den Ausgabedateinamen einfügen, wie wir zuvor dynamische Dateinamen zusammengestellt haben.

Nimm im process-Block die folgende Codeänderung vor:

NachherVorher

    output:
    path "COLLECTED-${batch_name}-output.txt"

    script:
    """
    cat ${input_files} > 'COLLECTED-${batch_name}-output.txt'
    """

    output:
    path "COLLECTED-output.txt"

    script:
    """
    cat ${input_files} > 'COLLECTED-output.txt'
    """

Dies richtet den process ein, den batch_name-Wert zu verwenden, um einen spezifischen Dateinamen für die endgültige Ausgabe des Workflows zu generieren.

3.2. Einen batch-Kommandozeilenparameter hinzufügen

Jetzt brauchen wir einen Weg, den Wert für batch_name zu liefern und ihn an den process-Aufruf zu übergeben.

3.2.1. params verwenden, um den Parameter einzurichten

Du weißt bereits, wie du das params-System verwendest, um CLI-Parameter zu deklarieren. Lass uns das verwenden, um einen batch-Parameter zu deklarieren (mit einem Standardwert, weil wir faul sind).

Nimm im Abschnitt für Pipeline-Parameter die folgenden Codeänderungen vor:

NachherVorher

/*
 * Pipeline-Parameter
 */
params {
    input: Path = 'data/greetings.csv'
    batch: String = 'batch'
}

/*
 * Pipeline-Parameter
 */
params {
    input: Path = 'data/greetings.csv'
}

Genau wie wir es für --input demonstriert haben, kannst du diesen Standardwert überschreiben, indem du einen Wert mit --batch auf der Kommandozeile angibst.

3.2.2. Den batch-Parameter an den process übergeben

Um den Wert des Parameters an den process zu übergeben, müssen wir ihn im process-Aufruf hinzufügen.

Nimm im workflow-Block die folgende Codeänderung vor:

NachherVorher

    // Alle Begrüßungen in einer Datei sammeln
    collectGreetings(convertToUpper.out.collect(), params.batch)

    // Alle Begrüßungen in einer Datei sammeln
    collectGreetings(convertToUpper.out.collect())

Du siehst, dass um mehrere Eingaben an einen process zu übergeben, du sie einfach in den Aufruf-Klammern auflistest, durch Kommas getrennt.

Warnung

Du MUSST die Eingaben an den process in der EXAKT GLEICHEN REIHENFOLGE übergeben, wie sie im Eingabedefinitionsblock des process aufgelistet sind.

3.3. Den Workflow ausführen

Lass uns versuchen, dies mit einem Stapelnamen auf der Kommandozeile auszuführen.

nextflow run hello-workflow.nf -resume --batch trio

Befehlsausgabe

N E X T F L O W   ~  version 25.10.2

Launching `hello-workflow.nf` [confident_rutherford] DSL2 - revision: bc58af409c

executor >  local (1)
[79/33b2f0] sayHello (2)       | 3 of 3, cached: 3 ✔
[99/79394f] convertToUpper (2) | 3 of 3, cached: 3 ✔
[b5/f19efe] collectGreetings   | 1 of 1 ✔

Es läuft erfolgreich und produziert die gewünschte Ausgabe:

Dateiinhalte

results/COLLECTED-trio-output.txt

HELLO
BONJOUR
HOLà

Solange wir den Parameter entsprechend angeben, werden nachfolgende Läufe auf anderen Eingabestapeln keine vorherigen Ergebnisse überschreiben.

Zusammenfassung

Du weißt, wie du mehr als eine Eingabe an einen process übergibst.

Was kommt als Nächstes?

Lerne, wie du mehrere Ausgaben ausgibst und sie bequem handhabst.

---

4. Eine Ausgabe zum Sammelschritt hinzufügen

Bisher haben wir processes verwendet, die jeweils nur eine Ausgabe produzierten. Wir konnten auf ihre jeweiligen Ausgaben sehr bequem mit der <process>.out-Syntax zugreifen, die wir sowohl im Kontext der Übergabe einer Ausgabe an den nächsten process (z.B. convertToUpper(sayHello.out)) als auch im Kontext des publish:-Abschnitts (z.B. first_output = sayHello.out) verwendet haben.

Was passiert, wenn ein process mehr als eine Ausgabe produziert? Wie handhaben wir die mehreren Ausgaben? Können wir eine bestimmte Ausgabe auswählen und verwenden?

Alles ausgezeichnete Fragen, und die kurze Antwort ist ja, das können wir!

Mehrere Ausgaben werden in separate channels verpackt. Wir können entweder wählen, diesen Ausgabe-channels Namen zu geben, was es einfach macht, sie später einzeln zu referenzieren, oder wir können sie per Index referenzieren.

Lass uns mit einem Beispiel eintauchen.

Zur Demonstration sagen wir, dass wir die Anzahl der Begrüßungen zählen wollen, die für einen gegebenen Eingabestapel gesammelt werden, und sie in einer Datei berichten wollen.

4.1. Den process so modifizieren, dass er die Anzahl der Begrüßungen zählt und ausgibt

Dies erfordert zwei wichtige Änderungen an der process-Definition: wir brauchen einen Weg, die Begrüßungen zu zählen und eine Berichtsdatei zu schreiben, dann müssen wir diese Berichtsdatei zum output-Block des process hinzufügen.

4.1.1. Die Anzahl der gesammelten Begrüßungen zählen

Praktischerweise erlaubt Nextflow uns, beliebigen Code im script:-Block der process-Definition hinzuzufügen, was wirklich praktisch ist, um solche Dinge zu tun.

Das bedeutet, wir können Nextflows eingebaute size()-Funktion verwenden, um die Anzahl der Dateien im input_files-Array zu erhalten, und das Ergebnis mit einem echo-Befehl in eine Datei schreiben.

Nimm im collectGreetings-process-Block die folgenden Codeänderungen vor:

NachherVorher

    script:
    count_greetings = input_files.size()
    """
    cat ${input_files} > 'COLLECTED-${batch_name}-output.txt'
    echo 'There were ${count_greetings} greetings in this batch.' > '${batch_name}-report.txt'
    """

    script:
    """
    cat ${input_files} > 'COLLECTED-${batch_name}-output.txt'
    """

Die count_greetings-Variable wird zur Laufzeit berechnet.

4.1.2. Die Berichtsdatei ausgeben und Ausgaben benennen

Im Prinzip müssen wir nur die Berichtsdatei zum output:-Block hinzufügen.

Aber während wir dabei sind, werden wir auch einige emit:-Tags zu unseren Ausgabedeklarationen hinzufügen. Diese werden es uns ermöglichen, die Ausgaben per Namen auszuwählen, anstatt Positionsindizes verwenden zu müssen.

Nimm im process-Block die folgende Codeänderung vor:

NachherVorher

    output:
    path "COLLECTED-${batch_name}-output.txt", emit: outfile
    path "${batch_name}-report.txt", emit: report

    output:
    path "COLLECTED-${batch_name}-output.txt"

Die emit:-Tags sind optional, und wir hätten ein Tag nur zu einer der Ausgaben hinzufügen können. Aber wie das Sprichwort sagt, warum nicht beides?

Tipp

Wenn du die Ausgaben eines process nicht mit emit: benennst, kannst du trotzdem einzeln auf sie zugreifen, indem du ihren jeweiligen (nullbasierten) Index verwendest. Zum Beispiel würdest du <process>.out[0] verwenden, um die erste Ausgabe zu bekommen, <process>.out[1] für die zweite Ausgabe, und so weiter.

Wir bevorzugen es, Ausgaben zu benennen, weil es sonst zu einfach ist, versehentlich den falschen Index zu greifen, besonders wenn der process viele Ausgaben produziert.

4.2. Die Workflow-Ausgaben aktualisieren

Jetzt, da wir zwei Ausgaben aus dem collectGreetings-process haben, enthält die collectGreetings.out-Ausgabe zwei channels:

• collectGreetings.out.outfile enthält die endgültige Ausgabedatei
• collectGreetings.out.report enthält die Berichtsdatei

Wir müssen die Workflow-Ausgaben entsprechend aktualisieren.

4.2.1. Den publish:-Abschnitt aktualisieren

Nimm im workflow-Block die folgende Codeänderung vor:

NachherVorher

    publish:
    first_output = sayHello.out
    uppercased = convertToUpper.out
    collected = collectGreetings.out.outfile
    batch_report = collectGreetings.out.report

    publish:
    first_output = sayHello.out
    uppercased = convertToUpper.out
    collected = collectGreetings.out

Wie du sehen kannst, ist das Referenzieren spezifischer process-Ausgaben jetzt trivial. Wenn wir in Teil 5 (Containers) einen weiteren Schritt zu unserer Pipeline hinzufügen, werden wir in der Lage sein, leicht auf collectGreetings.out.outfile zu verweisen und es dem neuen process zu übergeben (Spoiler: der neue process heißt cowpy).

Aber für jetzt aktualisieren wir die Workflow-Level-Ausgaben fertig.

4.2.2. Den output-Block aktualisieren

Nimm im output-Block die folgende Codeänderung vor:

NachherVorher

output {
    first_output {
        path 'hello_workflow'
        mode 'copy'
    }
    uppercased {
        path 'hello_workflow'
        mode 'copy'
    }
    collected {
        path 'hello_workflow'
        mode 'copy'
    }
    batch_report {
        path 'hello_workflow'
        mode 'copy'
    }
}

output {
    first_output {
        path 'hello_workflow'
        mode 'copy'
    }
    uppercased {
        path 'hello_workflow'
        mode 'copy'
    }
    collected {
        path 'hello_workflow'
        mode 'copy'
    }
}

Wir müssen die collected-Ausgabedefinition nicht aktualisieren, da sich dieser Name nicht geändert hat. Wir müssen nur die neue Ausgabe hinzufügen.

4.3. Den Workflow ausführen

Lass uns versuchen, dies mit dem aktuellen Stapel von Begrüßungen auszuführen.

nextflow run hello-workflow.nf -resume --batch trio

Befehlsausgabe

 N E X T F L O W   ~  version 25.10.2

Launching `hello-workflow.nf` [ecstatic_wilson] DSL2 - revision: c80285f8c8

executor >  local (1)
[c5/4c6ca9] sayHello (3)       [100%] 3 of 3, cached: 3 ✔
[0e/6cbc59] convertToUpper (3) [100%] 3 of 3, cached: 3 ✔
[02/61ead2] collectGreetings   [100%] 1 of 1 ✔

Wenn du ins Verzeichnis results/hello_workflow/ schaust, findest du die neue Berichtsdatei trio-report.txt. Öffne sie, um zu überprüfen, dass der Workflow korrekt die Anzahl der verarbeiteten Begrüßungen berichtet hat.

Dateiinhalte

trio-report.txt

There were 3 greetings in this batch.

Du kannst gerne weitere Begrüßungen zur CSV hinzufügen und testen, was passiert.

Zusammenfassung

Du weißt, wie du einen process mehrere benannte Ausgaben ausgeben lässt und wie du sie auf Workflow-Ebene angemessen handhabst.

Allgemeiner verstehst du die wichtigsten Prinzipien, die beim Verbinden von processes auf gängige Weise involviert sind.

Was kommt als Nächstes?

Mach eine extra lange Pause, du hast sie verdient.

Wenn du bereit bist, gehe zu Teil 4: Hallo Module, um zu lernen, wie du deinen Code für bessere Wartbarkeit und Code-Effizienz modularisierst.

---

Quiz

#

Wie greifst du auf die Ausgabe eines process im workflow-Block zu?

process.output

output.processName

processName.out

get(processName)

Mehr erfahren: 1.4. Die Ausgabe des ersten process an den zweiten process übergeben

#

Was bestimmt die Reihenfolge der process-Ausführung in Nextflow?

Die Reihenfolge, in der processes im workflow-Block geschrieben sind

Alphabetische Reihenfolge nach process-Namen

Datenabhängigkeiten zwischen processes

Zufällige Reihenfolge für parallele Ausführung

Mehr erfahren: 1.4. Die Ausgabe des ersten process an den zweiten process übergeben

#

Welcher Operator sollte ??? ersetzen, um alle Ausgaben in eine einzelne Liste für den nachfolgenden process zu sammeln?

workflow {
    greetings_ch = Channel.of('Hello', 'Bonjour', 'Hola')
    SAYHELLO(greetings_ch)
    GATHER_ALL(SAYHELLO.out.???)
}

flatten()

collect()

mix()

join()

Mehr erfahren: 2.4. Einen Operator verwenden, um die Begrüßungen in eine einzelne Eingabe zu sammeln

#

Wann solltest du den collect()-Operator verwenden?

Wenn du Elemente parallel verarbeiten möchtest

Wenn du channel-Inhalte filtern musst

Wenn ein nachfolgender process alle Elemente von einem vorgelagerten process benötigt

Wenn du Daten auf mehrere processes aufteilen möchtest

Mehr erfahren: 2.4. Einen Operator verwenden, um die Begrüßungen in eine einzelne Eingabe zu sammeln

#

Wie greifst du auf eine benannte Ausgabe von einem process zu?

processName.outputName

processName.get(outputName)

processName.out.outputName

output.processName.outputName

Mehr erfahren: 4.1.2. Die Berichtsdatei ausgeben und Ausgaben benennen

#

Was ist die korrekte Syntax, um eine Ausgabe in einem process zu benennen?

name: outputName

output: outputName

emit: outputName

label: outputName

Mehr erfahren: 4.1.2. Die Berichtsdatei ausgeben und Ausgaben benennen

#

Was muss bei der Übergabe mehrerer Eingaben an einen process zutreffen?

Alle Eingaben müssen vom gleichen Typ sein

Eingaben müssen in alphabetischer Reihenfolge übergeben werden

Die Reihenfolge der Eingaben muss mit der im Eingabeblock definierten Reihenfolge übereinstimmen

Nur zwei Eingaben können gleichzeitig übergeben werden

Mehr erfahren: 3. Mehr als eine Eingabe an einen process übergeben