# Stream CSV Reader

O **Stream CSV Reader** lê um arquivo CSV local linha por linha em uma estrutura JSON e aciona subfluxos para processar cada linha. Esse recurso é recomendado para cenários em que arquivos grandes precisam ser processados de forma eficiente e em escala.

## **Parâmetros**

Configure o conector usando os parâmetros abaixo. Os campos que suportam [expressões Double Braces](https://docs.digibee.com/documentation/connectors-and-triggers/pt-br/double-braces) são marcados na coluna Suporta DB.

{% hint style="info" %}
Este conector é compatível com o parâmetro **Alias**, que permite salvar sua saída e referenciá-la posteriormente no fluxo usando expressões Double Braces. [Saiba mais](https://docs.digibee.com/documentation/connectors-and-triggers/pt-br/double-braces/how-to-reference-data-using-double-braces/previous-steps-access).
{% endhint %}

{% tabs fullWidth="true" %}
{% tab title="General" %}

<table data-full-width="true"><thead><tr><th>Parâmetro</th><th>Descrição</th><th>Tipo</th><th>Suporta DB</th><th>Padrão</th></tr></thead><tbody><tr><td><strong>File Name</strong></td><td>Nome do arquivo CSV local a ser lido.</td><td>String</td><td>✅</td><td><code>data.csv</code></td></tr><tr><td><strong>Charset</strong></td><td>Codificação de caracteres usada para ler o arquivo.</td><td>String</td><td>❌</td><td><code>UTF-8</code></td></tr><tr><td><strong>Element Identifier</strong></td><td>Em caso de erros, os atributos definidos serão enviados para o subprocesso onException.</td><td>String</td><td>❌</td><td>data</td></tr><tr><td><strong>Parallel Execution Of Each Iteration</strong></td><td>Se ativado, cada linha é processada em paralelo.</td><td>Booleano</td><td>❌</td><td>False</td></tr><tr><td><strong>Ignore Invalid Charset</strong></td><td>Se ativado, caracteres com codificação inválida são ignorados.</td><td>Booleano</td><td>❌</td><td>False</td></tr><tr><td><strong>Fail On Error</strong></td><td>Se ativado, a execução do pipeline é interrompida em caso de erro. Se desativado, a execução continua, mas a propriedade <code>"success"</code> será definida como <code>false</code>.</td><td>Booleano</td><td>❌</td><td>False</td></tr><tr><td><strong>Advanced</strong></td><td>Ativa parâmetros avançados.</td><td>Booleano</td><td>❌</td><td>False</td></tr><tr><td><strong>Delimiter</strong></td><td>Define qual delimitador usar.</td><td>String</td><td>❌</td><td>N/A</td></tr><tr><td><strong>Skip</strong></td><td>O número de linhas a serem puladas antes de começar a ler o arquivo.</td><td>Integer</td><td>✅</td><td>N/A</td></tr><tr><td><strong>Limit</strong></td><td>Número máximo de linhas a serem lidas do arquivo. Um valor igual a 0 indica que não há limite.</td><td>Integer</td><td>✅</td><td><code>0</code></td></tr><tr><td><strong>Chunk Size</strong></td><td>Number of lines to process in each batch.</td><td>Integer</td><td>✅</td><td>1</td></tr><tr><td><strong>Ignore Header</strong></td><td>Se ativado, a primeira linha (cabeçalho) do arquivo é ignorada.</td><td>Booleano</td><td>❌</td><td>False</td></tr><tr><td><strong>Custom Header (,)</strong></td><td>Define uma lista separada por vírgulas de cabeçalhos personalizados a serem usados ​​em vez dos cabeçalhos CSV originais.</td><td>Integer</td><td>❌</td><td>N/A</td></tr><tr><td><strong>Filter Headers (,)</strong></td><td>Define quais cabeçalhos CSV devem ser lidos, fornecidos como uma lista separada por vírgulas.</td><td>Integer</td><td>❌</td><td>N/A</td></tr></tbody></table>

{% endtab %}

{% tab title="Documentation" %}

<table data-full-width="true"><thead><tr><th>Parâmetro</th><th>Descrição</th><th>Valor padrão</th><th>Tipo de dado</th></tr></thead><tbody><tr><td><strong>Documentation</strong></td><td>Campo opcional para descrever a configuração do conector e quaisquer regras de negócio relevantes.</td><td>N/A</td><td>String</td></tr></tbody></table>
{% endtab %}
{% endtabs %}

## **Fluxo de mensagens**

### **Entrada**

O conector aguarda uma mensagem no seguinte formato:

```json
{
  "filename": "fileName"
}
```

### **Saída**

```json
{
  "total": 0,
  "success": 0,
  "failed": 0
}
```

* `total`: Número total de linhas processadas.
* `success`: Número total de linhas processadas com sucesso.
* `failed`: Número total de linhas cujo processamento falhou.

{% hint style="info" %}
Para informar que uma linha foi processada corretamente e incrementar o campo `"success"`, o subpipeline onProcess deve retornar `{ "success": true }` ao final de cada execução. Essa é a única forma de o output refletir corretamente o resultado dos processamentos.
{% endhint %}

## Definindo os subpipelines

Para definir os subpipelines a serem executados em cada iteração, clique nos ícones **onProcess** ou **onException** no conector Stream CSV Reader. Ao clicar em um desses ícones, o subpipeline correspondente será criado (ou exibido, caso já exista).

### Acessando dados dentro do subpipeline onProcess

Os dados disponíveis dentro do subpipeline **onProcess** dependem da configuração do parâmetro **Chunk Size**.

#### Modo linha a linha (Chunk Size = 1)

Quando o **Chunk Size** é `1` (padrão), cada linha do CSV é convertida em um objeto JSON e enviada individualmente ao subpipeline **onProcess**. As chaves do objeto correspondem aos cabeçalhos do CSV (ou aos **Custom Headers**, se configurados).

Dado um arquivo CSV com o seguinte conteúdo:

```
id,nome,email
1,João Silva,joao@exemplo.com
2,Maria Souza,maria@exemplo.com
```

Cada chamada ao subpipeline **onProcess** recebe um corpo no seguinte formato:

```json
{
    "id": "1",
    "nome": "João Silva",
    "email": "joao@exemplo.com"
}
```

Você pode acessar qualquer campo da linha diretamente usando expressões Double Braces. Por exemplo, para referenciar a coluna `nome`:

```
{{$.nome}}
```

Se o parâmetro **Element Identifier** estiver configurado, um campo `processedId` contendo o índice da linha também é incluído no corpo da mensagem:

<pre class="language-json"><code class="lang-json">{
    "id": "1",
<strong>    "nome": "João Silva",
</strong>    "email": "joao@exemplo.com",
    "processedId": "1"
}
</code></pre>

#### Modo lote (Chunk Size > 1)

Quando o **Chunk Size** é maior que `1`, as linhas são agrupadas em lotes antes de serem enviadas ao subpipeline **onProcess**. O corpo da mensagem recebida contém o lote como um array sob a chave `chunk`:

```json
{
    "chunk": [
        { "id": "1", "nome": "João Silva", "email": "joao@exemplo.com" },
        { "id": "2", "nome": "Maria Souza", "email": "maria@exemplo.com" }
    ],
    "chunkSize": 2,
    "startIndex": 0
}
```

<table data-full-width="false"><thead><tr><th>Campo</th><th>Descrição</th></tr></thead><tbody><tr><td><code>chunk</code></td><td>Array de objetos JSON, um por linha do CSV no lote.</td></tr><tr><td><code>chunkSize</code></td><td>Número de linhas no lote atual.</td></tr><tr><td><code>startIndex</code></td><td>Índice da primeira linha do lote no arquivo original.</td></tr></tbody></table>

## Tratamento de erros no loop

Quando ocorre um erro durante o processamento de uma linha ou lote, o conector aciona o subpipeline **onException**, se configurado. O corpo da mensagem enviada depende do modo de processamento:

#### Erro em modo linha a linha

```json
{
    "processedId": "1",
    "error": "mensagem de erro",
    "exception": "java.io.IOException"
}
```

| Campo         | Descrição                                |
| ------------- | ---------------------------------------- |
| `processedId` | Índice da linha que falhou.              |
| `error`       | Mensagem de erro retornada pela exceção. |
| `exception`   | Nome completo da classe da exceção.      |

#### Erro em modo lote

```json
{
    "startIndex": 10,
    "chunkSize": 5,
    "error": "mensagem de erro",
    "exception": "java.lang.Exception"
}
```

| **Campo**    | **Descrição**                                |
| ------------ | -------------------------------------------- |
| `startIndex` | Índice da primeira linha do lote que falhou. |
| `chunkSize`  | Número de linhas no lote que falhou.         |
| `error`      | Mensagem de erro retornada pela exceção.     |
| `exception`  | Nome completo da classe da exceção.          |

{% hint style="warning" %}
Se ocorrer um erro estrutural grave dentro do próprio subpipeline **onException**, a execução do pipeline será interrompida e o erro será propagado para o próximo conector no fluxo principal.
{% endhint %}

## **Informações adicionais**

* O conector lança uma exceção se o **File Name** não existir ou não puder ser lido.
* A manipulação de arquivos dentro de um pipeline ocorre de forma protegida. Todos os arquivos são acessados por meio de um diretório temporário, no qual cada chave de pipeline tem acesso apenas ao seu próprio conjunto de arquivos.
* Este conector realiza o processamento em lote, o que significa que os dados são processados de forma contínua, controlada e em pequenos lotes.