7. Canais¶

Canais são uma estrutura de dados chave do Nextflow que permite a implementação de fluxos de trabalho computacionais utilizando paradigmas funcional e reativo com base no paradigma de programação Dataflow.

Eles são usados para conectar logicamente tarefas entre si ou para implementar transformações de dados de estilo funcional.

7.1 Tipos de canais¶

O Nextflow distingue dois tipos diferentes de canais: canais de fila e canais de valor.

7.1.1 Canal de fila¶

Um canal de fila é uma fila assíncrona unidirecional FIFO (First-in-First-out, o primeiro a entrar, é o primeiro a sair) que conecta dois processos ou operadores.

• assíncrono significa que as operações ocorrem sem bloqueio.
• unidirecional significa que os dados fluem do gerador para o consumidor.
• FIFO significa que os dados são entregues na mesma ordem em que são produzidos. Primeiro a entrar, primeiro a sair.

Um canal de fila é criado implicitamente por definições de saída de um processo ou usando fábricas de canal, como o Channel.of ou Channel.fromPath.

Tente os seguintes trechos de código:

1
2
3

canal = Channel.of(1, 2, 3)
println(canal) // (1)!
canal.view() // (2)!

1. Use a função println embutida no Nextflow por padrão para imprimir o conteúdo do canal canal
2. Aplique o operador view no canal canal para imprimir cada emissão desse canal

1
2

canal = Channel.of(1, 2, 3)
canal.view()

7.1.2 Canais de valor¶

Um canal de valor (também conhecido como canal singleton), por definição, está vinculado a um único valor e pode ser lido quantas vezes for necessário sem consumir seu conteúdo. Um canal de valor é criado usando a fábrica de canal value ou por operadores que retornam um valor apenas, como first, last, collect, count, min, max, reduce, e sum.

Para entender melhor a diferença entre canais de valor e de fila, salve o trecho abaixo como exemplo.nf.

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20

canal1 = Channel.of(1, 2, 3)
canal2 = Channel.of(1)

process SUM {
    input:
    val x
    val y

    output:
    stdout

    script:
    """
    echo \$(($x+$y))
    """
}

workflow {
    SUM(canal1, canal2).view()
}

Ao rodar o script, ele imprime apenas 2, como você pode ver abaixo:

2

Um processo só instanciará uma tarefa quando houver elementos a serem consumidos de todos os canais fornecidos como entrada para ele. Como canal1 e canal2 são canais de fila, e o único elemento de canal2 foi consumido, nenhuma nova instância de processo será iniciada, mesmo se houver outros elementos a serem consumidos em canal1.

Para usar o único elemento em canal2 várias vezes, podemos usar Channel.value como mencionado acima, ou usar um operador de canal que retorna um único elemento como first() abaixo:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20

canal1 = Channel.of(1, 2, 3)
canal2 = Channel.of(1)

process SUM {
    input:
    val x
    val y

    output:
    stdout

    script:
    """
    echo \$(($x+$y))
    """
}

workflow {
    SUM(canal1, canal2.first()).view()
}

4

3

2

Além disso, em muitas situações, o Nextflow converterá implicitamente variáveis em canais de valor quando forem usadas em uma chamada de processo. Por exemplo, quando você chama um processo com um parâmetro de fluxo de trabalho (params.exemplo) que possui um valor de string, ele é automaticamente convertido em um canal de valor.

7.2 Fábricas de canal¶

Estes são comandos do Nextflow para criar canais que possuem entradas e funções implícitas esperadas.

7.2.1 value()¶

A fábrica de canal value é utilizada para criar um canal de valor. Um argumento opcional não nulo pode ser especificado para vincular o canal a um valor específico. Por exemplo:

1
2
3

canal1 = Channel.value() // (1)!
canal2 = Channel.value('Olá, você!') // (2)!
canal3 = Channel.value([1, 2, 3, 4, 5]) // (3)!

1. Cria um canal de valor vazio
2. Cria um canal de valor e vincula uma string a ele
3. Cria um canal de valor e vincula a ele um objeto de lista que será emitido como uma única emissão

7.2.2 of()¶

A fábrica Channel.of permite a criação de um canal de fila com os valores especificados como argumentos.

1
2

canal = Channel.of(1, 3, 5, 7)
canal.view { "numero: $it" }

A primeira linha neste exemplo cria uma variável canal que contém um objeto de canal. Este canal emite os valores especificados como parâmetro na fábrica de canal of. Assim, a segunda linha imprimirá o seguinte:

numero: 1
numero: 3
numero: 5
numero: 7

A fábrica de canal Channel.of funciona de maneira semelhante ao Channel.from (que foi descontinuado), corrigindo alguns comportamentos inconsistentes do último e fornecendo um melhor manuseio quando um intervalo de valores é especificado. Por exemplo, o seguinte funciona com um intervalo de 1 a 23:

1
2
3

Channel
    .of(1..23, 'X', 'Y')
    .view()

7.2.3 fromList()¶

A fábrica de canal Channel.fromList cria um canal emitindo os elementos fornecidos por um objeto de lista especificado como um argumento:

1
2
3
4
5

list = ['olá', 'mundo']

Channel
    .fromList(list)
    .view()

7.2.4 fromPath()¶

A fábrica de canal fromPath cria um canal de fila emitindo um ou mais arquivos correspondentes ao padrão glob especificado.

1

Channel.fromPath('./data/meta/*.csv')

Este exemplo cria um canal e emite tantos itens quanto arquivos com extensão csv existirem na pasta ./data/meta. Cada elemento é um objeto de arquivo implementando a interface Path do Java.

Saiba mais sobre a sintaxe dos padrões glob neste link.

1
2
3

Channel
    .fromPath('./data/ggal/**.fq', hidden: true)
    .view()

7.2.5 fromFilePairs()¶

A fábrica de canal fromFilePairs cria um canal emitindo os pares de arquivos correspondentes a um padrão glob fornecido pelo usuário. Os arquivos correspondentes são emitidos como tuplas, nas quais o primeiro elemento é a chave de agrupamento do par correspondente e o segundo elemento é a lista de arquivos (classificados em ordem lexicográfica).

1
2
3

Channel
    .fromFilePairs('./data/ggal/*_{1,2}.fq')
    .view()

Ele produzirá uma saída semelhante à seguinte:

[liver, [/workspace/gitpod/nf-training/data/ggal/liver_1.fq, /workspace/gitpod/nf-training/data/ggal/liver_2.fq]]
[gut, [/workspace/gitpod/nf-training/data/ggal/gut_1.fq, /workspace/gitpod/nf-training/data/ggal/gut_2.fq]]
[lung, [/workspace/gitpod/nf-training/data/ggal/lung_1.fq, /workspace/gitpod/nf-training/data/ggal/lung_2.fq]]

1
2
3

Channel
    .fromFilePairs('./data/ggal/*_{1,2}.fq', flat: true)
    .view()

7.2.6 fromSRA()¶

A fábrica de canal Channel.fromSRA permite consultar o banco de dados NCBI SRA e retorna um canal que emite os arquivos FASTQ correspondentes aos critérios de seleção especificados.

A consulta pode ser ID(s) de projeto(s) ou número(s) de acesso suportado(s) pela API do NCBI ESearch.

Por exemplo, o trecho a seguir imprimirá o conteúdo de um ID de projeto NCBI:

1
2
3
4
5

params.ncbi_api_key = '<Sua chave da API aqui>'

Channel
    .fromSRA(['SRP073307'], apiKey: params.ncbi_api_key)
    .view()

Isso deve imprimir:

[SRR3383346, [/vol1/fastq/SRR338/006/SRR3383346/SRR3383346_1.fastq.gz, /vol1/fastq/SRR338/006/SRR3383346/SRR3383346_2.fastq.gz]]
[SRR3383347, [/vol1/fastq/SRR338/007/SRR3383347/SRR3383347_1.fastq.gz, /vol1/fastq/SRR338/007/SRR3383347/SRR3383347_2.fastq.gz]]
[SRR3383344, [/vol1/fastq/SRR338/004/SRR3383344/SRR3383344_1.fastq.gz, /vol1/fastq/SRR338/004/SRR3383344/SRR3383344_2.fastq.gz]]
[SRR3383345, [/vol1/fastq/SRR338/005/SRR3383345/SRR3383345_1.fastq.gz, /vol1/fastq/SRR338/005/SRR3383345/SRR3383345_2.fastq.gz]]
// (o restante foi omitido)

Vários IDs de acesso podem ser especificados usando um objeto lista:

1
2
3
4

ids = ['ERR908507', 'ERR908506', 'ERR908505']
Channel
    .fromSRA(ids, apiKey: params.ncbi_api_key)
    .view()

[ERR908507, [/vol1/fastq/ERR908/ERR908507/ERR908507_1.fastq.gz, /vol1/fastq/ERR908/ERR908507/ERR908507_2.fastq.gz]]
[ERR908506, [/vol1/fastq/ERR908/ERR908506/ERR908506_1.fastq.gz, /vol1/fastq/ERR908/ERR908506/ERR908506_2.fastq.gz]]
[ERR908505, [/vol1/fastq/ERR908/ERR908505/ERR908505_1.fastq.gz, /vol1/fastq/ERR908/ERR908505/ERR908505_2.fastq.gz]]

É fácil usar este canal como uma entrada usando a sintaxe usual do Nextflow. O código abaixo cria um canal contendo duas amostras de um estudo SRA público e executa o FASTQC nos arquivos resultantes. Veja:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22

params.ncbi_chave_api = '<Sua chave de API aqui>'

params.accession = ['ERR908507', 'ERR908506']

process FASTQC {
    input:
    tuple val(id_amostra), path(arquivo_de_leituras)

    output:
    path("fastqc_${id_amostra}_logs")

    script:
    """
    mkdir fastqc_${id_amostra}_logs
    fastqc -o fastqc_${id_amostra}_logs -f fastq -q ${arquivo_de_leituras}
    """
}

workflow {
    leituras = Channel.fromSRA(params.accession, apiKey: params.ncbi_chave_api)
    FASTQC(leituras)
}

Se você deseja executar o fluxo de trabalho acima e não possui o fastqc instalado em sua máquina, não esqueça o que aprendeu na seção anterior. Execute este fluxo de trabalho com -with-docker biocontainers/fastqc:v0.11.5, por exemplo.

7.2.7 Arquivos de texto¶

O operador splitText permite dividir strings de várias linhas ou itens de arquivo de texto, emitidos por um canal de origem em blocos contendo n linhas, que serão emitidos pelo canal resultante. Veja:

1
2
3
4

Channel
    .fromPath('data/meta/random.txt') // (1)!
    .splitText() // (2)!
    .view() // (3)!

1. Instrui o Nextflow a criar um canal a partir do caminho data/meta/random.txt
2. O operador splitText divide cada item em pedaços de uma linha por padrão.
3. Veja o conteúdo do canal.

Você pode definir o número de linhas em cada bloco usando o parâmetro by, conforme mostrado no exemplo a seguir:

1
2
3
4
5
6
7

Channel
    .fromPath('data/meta/random.txt')
    .splitText(by: 2)
    .subscribe {
        print it;
        print "--- fim do bloco ---\n"
    }

Uma clausura opcional pode ser especificada para transformar os blocos de texto produzidos pelo operador. O exemplo a seguir mostra como dividir arquivos de texto em blocos de 10 linhas e transformá-los em letras maiúsculas:

1
2
3
4

Channel
    .fromPath('data/meta/random.txt')
    .splitText(by: 10) { it.toUpperCase() }
    .view()

Você também pode fazer contagens para cada linha:

1
2
3
4
5
6

contador = 0

Channel
    .fromPath('data/meta/random.txt')
    .splitText()
    .view { "${contador++}: ${it.toUpperCase().trim()}" }

Por fim, você também pode usar o operador em arquivos simples (fora do contexto do canal):

1
2
3
4
5
6

def f = file('data/meta/random.txt')
def linhas = f.splitText()
def contador = 0
for (String linha : linhas) {
    log.info "${contador++} ${linha.toUpperCase()}"
}

7.2.8 Valores separados por vírgula (.csv)¶

O operador splitCsv permite analisar itens de texto formatados em CSV (Comma-separated value) emitidos por um canal.

Em seguida, ele os divide em registros ou os agrupa como uma lista de registros com um comprimento especificado.

No caso mais simples, basta aplicar o operador splitCsv a um canal que emite arquivos de texto ou entradas de texto no formato CSV. Por exemplo, para visualizar apenas a primeira e a quarta colunas:

1
2
3
4
5

Channel
    .fromPath("data/meta/patients_1.csv")
    .splitCsv()
    // linha é um objeto de lista
    .view { linha -> "${linha[0]}, ${linha[3]}" }

Quando o CSV começa com uma linha de cabeçalho definindo os nomes das colunas, você pode especificar o parâmetro header: true que permite referenciar cada valor pelo nome da coluna, conforme mostrado no exemplo a seguir:

1
2
3
4
5

Channel
    .fromPath("data/meta/patients_1.csv")
    .splitCsv(header: true)
    // linha é um objeto de lista
    .view { linha -> "${linha.patient_id}, ${linha.num_samples}" }

Como alternativa, você pode fornecer nomes de cabeçalho personalizados especificando uma lista de strings no parâmetro de cabeçalho, conforme mostrado abaixo:

1
2
3
4
5

Channel
    .fromPath("data/meta/patients_1.csv")
    .splitCsv(header: ['col1', 'col2', 'col3', 'col4', 'col5'])
    // linha é um objeto de lista
    .view { linha -> "${linha.col1}, ${linha.col4}" }

Você também pode processar vários arquivos CSV ao mesmo tempo:

1
2
3
4

Channel
    .fromPath("data/meta/patients_*.csv") // <-- use um padrão de captura
    .splitCsv(header: true)
    .view { linha -> "${linha.patient_id}\t${linha.num_samples}" }

Por fim, você também pode operar em arquivos CSV fora do contexto do canal:

1
2
3
4
5

def f = file('data/meta/patients_1.csv')
def linhas = f.splitCsv()
for (List linha : linhas) {
    log.info "${linha[0]} -- ${linha[2]}"
}

gut,/workspace/gitpod/nf-training/data/ggal/gut_1.fq,/workspace/gitpod/nf-training/data/ggal/gut_2.fq

1
2
3

Channel
    .fromFilePairs(params.reads, checkIfExists: true)
    .set { read_pairs_ch }

1
2
3
4
5

Channel
    .fromPath("fastq.csv")
    .splitCsv()
    .view { linha -> "${linha[0]}, ${linha[1]}, ${linha[2]}" }
    .set { read_pairs_ch }

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15

process QUANTIFICATION {
    tag "$sample_id"

    input:
    path salmon_index
    tuple val(sample_id), path(reads)

    output:
    path sample_id, emit: quant_ch

    script:
    """
    salmon quant --threads $task.cpus --libType=U -i $salmon_index -1 ${reads[0]} -2 ${reads[1]} -o $sample_id
    """
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15

process QUANTIFICATION {
    tag "$sample_id"

    input:
    path salmon_index
    tuple val(sample_id), path(reads1), path(reads2)

    output:
    path sample_id, emit: quant_ch

    script:
    """
    salmon quant --threads $task.cpus --libType=U -i $salmon_index -1 ${reads1} -2 ${reads2} -o $sample_id
    """
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15

process FASTQC {
    tag "FASTQC on $sample_id"

    input:
    tuple val(sample_id), path(reads1), path(reads2)

    output:
    path "fastqc_${sample_id}_logs"

    script:
    """
    mkdir fastqc_${sample_id}_logs
    fastqc -o fastqc_${sample_id}_logs -f fastq -q ${reads1} ${reads2}
    """
}

7.2.9 Valores separados por tabulação (.tsv)¶

A análise de arquivos TSV funciona de maneira semelhante, basta adicionar a opção sep: '\t' no contexto do splitCsv:

1
2
3
4
5

Channel
    .fromPath("data/meta/regions.tsv", checkIfExists: true)
    // Use a opção `sep` para analisar arquivos com tabulação como separador
    .splitCsv(sep: '\t')
    .view()

1
2
3
4
5
6

Channel
    .fromPath("data/meta/regions.tsv", checkIfExists: true)
    // Use a opção `sep` para analisar arquivos com tabulação como separador
    .splitCsv(sep: '\t', header: true)
    // linha é um objeto de lista
    .view { linha -> "${linha.patient_id}" }

7.3 Formatos de arquivo mais complexos¶

7.3.1 JSON¶

Também podemos analisar facilmente o formato de arquivo JSON usando o oeprador de canal splitJson.

O operador splitJson suporta arranjos JSON:

1
2
3
4

Channel
    .of('["Domingo", "Segunda", "Terça", "Quarta", "Quinta", "Sexta", "Sábado"]')
    .splitJson()
    .view { "Item: ${it}" }

Item: Domingo
Item: Segunda
Item: Terça
Item: Quarta
Item: Quinta
Item: Sexta
Item: Sábado

Objetos JSON:

1
2
3
4

Channel
    .of('{"jogador": {"nome": "Bob", "altura": 180, "venceu_campeonato": false}}')
    .splitJson()
    .view { "Item: ${it}" }

Item: [key:jogador, value:[nome:Bob, altura:180, venceu_campeonato:false]]

E inclusive arranjos JSON com objetos JSON!

1
2
3
4
5

Channel
    .of('[{"nome": "Bob", "altura": 180, "venceu_campeonato": false}, \
        {"nome": "Alice", "height": 170, "venceu_campeonato": false}]')
    .splitJson()
    .view { "Item: ${it}" }

Item: [nome:Bob, altura:180, venceu_campeonato:false]
Item: [nome:Alice, altura:170, venceu_campeonato:false]

Arquivos contendo dados em formato JSON também podem ser analisados:

1
2
3
4

Channel
    .fromPath('arquivo.json')
    .splitJson()
    .view { "Item: ${it}" }

[{"nome": "Bob", "altura": 180, "venceu_campeonato": false}, {"nome": "Alice", "altura": 170, "venceu_campeonato": false}]

Item: [nome:Bob, altura:180, venceu_campeonato:false]
Item: [nome:Alice, altura:170, venceu_campeonato:false]

7.3.2 YAML¶

Isso também pode ser usado como uma forma de analisar arquivos YAML:

1
2
3
4
5
6
7
8

import org.yaml.snakeyaml.Yaml

def f = file('data/meta/regions.yml')
def registros = new Yaml().load(f)

for (def entrada : registros) {
    log.info "$entrada.patient_id -- $entrada.feature"
}

- patient_id: ATX-TBL-001-GB-01-105
  region_id: R1
  feature: pass_vafqc_flag
  pass_flag: "TRUE"
- patient_id: ATX-TBL-001-GB-01-105
  region_id: R1
  feature: pass_stripy_flag
  pass_flag: "TRUE"
- patient_id: ATX-TBL-001-GB-01-105
  region_id: R1
  feature: pass_manual_flag
  pass_flag: "TRUE"
- patient_id: ATX-TBL-001-GB-01-105
  region_id: R1
  feature: other_region_selection_flag
  pass_flag: "TRUE"
- patient_id: ATX-TBL-001-GB-01-105
  region_id: R1
  feature: ace_information_gained
  pass_flag: "TRUE"
- patient_id: ATX-TBL-001-GB-01-105
  region_id: R1
  feature: concordance_flag
  pass_flag: "TRUE"
- patient_id: ATX-TBL-001-GB-01-105
  region_id: R2
  feature: pass_vafqc_flag
  pass_flag: "TRUE"
- patient_id: ATX-TBL-001-GB-01-105
  region_id: R2
  feature: pass_stripy_flag
  pass_flag: "TRUE"
- patient_id: ATX-TBL-001-GB-01-105
  region_id: R2
  feature: pass_manual_flag
  pass_flag: "TRUE"
- patient_id: ATX-TBL-001-GB-01-105
  region_id: R2
  feature: other_region_selection_flag
  pass_flag: "TRUE"
- patient_id: ATX-TBL-001-GB-01-105
  region_id: R2
  feature: ace_information_gained
  pass_flag: "TRUE"
- patient_id: ATX-TBL-001-GB-01-105
  region_id: R2
  feature: concordance_flag
  pass_flag: "TRUE"
- patient_id: ATX-TBL-001-GB-01-105
  region_id: R3
  feature: pass_vafqc_flag
  pass_flag: "TRUE"
- patient_id: ATX-TBL-001-GB-01-105
  region_id: R3
  feature: pass_stripy_flag
  pass_flag: "FALSE"

ATX-TBL-001-GB-01-105 -- pass_vafqc_flag
ATX-TBL-001-GB-01-105 -- pass_stripy_flag
ATX-TBL-001-GB-01-105 -- pass_manual_flag
ATX-TBL-001-GB-01-105 -- other_region_selection_flag
ATX-TBL-001-GB-01-105 -- ace_information_gained
ATX-TBL-001-GB-01-105 -- concordance_flag
ATX-TBL-001-GB-01-105 -- pass_vafqc_flag
ATX-TBL-001-GB-01-105 -- pass_stripy_flag
ATX-TBL-001-GB-01-105 -- pass_manual_flag
ATX-TBL-001-GB-01-105 -- other_region_selection_flag
ATX-TBL-001-GB-01-105 -- ace_information_gained
ATX-TBL-001-GB-01-105 -- concordance_flag
ATX-TBL-001-GB-01-105 -- pass_vafqc_flag
ATX-TBL-001-GB-01-105 -- pass_stripy_flag

7.3.3 Armazenamento em módulos de analisadores sintáticos¶

A melhor maneira de armazenar scripts com analisadores é mantê-los em um arquivo de módulo Nextflow.

Digamos que não temos um operador de canal JSON, mas criamos uma função para isso. O arquivo parsers.nf deve conter a função parseArquivoJson. Veja o conteúdo abaixo:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24

include { parseArquivoJson } from './modules/parsers.nf'

process FOO {
    input:
    tuple val(id_paciente), val(caracteristica)

    output:
    stdout

    script:
    """
    echo $id_paciente tem $caracteristica como coluna
    """
}

workflow {
    Channel
        .fromPath('data/meta/regions*.json')
        | flatMap { parseArquivoJson(it) }
        | map { registro -> [registro.patient_id, registro.feature] }
        | unique
        | FOO
        | view
}

1
2
3
4
5
6
7

import groovy.json.JsonSlurper

def parseArquivoJson(arquivo_json) {
    def f = file(arquivo_json)
    def registros = new JsonSlurper().parse(f)
    return registros
}

ATX-TBL-001-GB-01-105 tem pass_stripy_flag como coluna

ATX-TBL-001-GB-01-105 tem ace_information_gained como coluna

ATX-TBL-001-GB-01-105 tem concordance_flag como coluna

ATX-TBL-001-GB-01-105 tem pass_vafqc_flag como coluna

ATX-TBL-001-GB-01-105 tem pass_manual_flag como coluna

ATX-TBL-001-GB-01-105 tem other_region_selection_flag como coluna

O Nextflow usará isso como uma função personalizada dentro do escopo workflow.