Como fazer limpeza incremental de ficheiros CSV em Azure Data Factory

Este tutorial explica como implementar uma pipeline em Azure Data Factory para detetar ficheiros CSV novos ou modificados, validar/limpar os dados e carregar apenas os registos incrementais para uma tabela em Azure SQL. É útil para reduzir custos e evitar duplicação ao processar ingestões recorrentes de ficheiros.

Pré-requisitos

• Conta Azure com permissões para criar recursos.
• Um recurso Azure Data Factory v2 criado.
• Armazenamento Blob ou ADLS Gen2 com alguns ficheiros CSV de exemplo.
• Uma base de dados Azure SQL com tabela alvo e acesso de escrita.
• Conhecimentos básicos de pipelines, datasets e activities no Azure Data Factory.

Passo 1: Conceito — como funciona a ingestão incremental

Explicação simples: usamos a lista de ficheiros (Get Metadata) para detetar ficheiros novos/alterados com base em lastModified. Guardamos um registo do último processamento (watermark) e só processamos ficheiros com lastModified posterior. Isto evita reprocessar ficheiros já validados.

Passo 2: Criar um ficheiro de controlo (watermark) em Blob

Vamos guardar a data/hora do último processamento num pequeno ficheiro JSON no blob. Crie um ficheiro chamado watermark.json com conteúdo inicial:

{
  "lastRun":"1970-01-01T00:00:00Z"
}

Passo 3: Pipeline — obter lista de ficheiros novos

No ADF crie um pipeline com as seguintes actividades: Get Metadata para listar ficheiros, Lookup para ler o watermark.json, Filter para seleccionar apenas os ficheiros com lastModified > watermark.

// Get Metadata dataset: apontar para a pasta de CSVs, field: childItems
// Lookup dataset: apontar para watermark.json
// Exemplo de expressão no Filter activity para comparar datas
@greater(item().lastModified, pipeline().parameters.lastWatermark)

Passo 4: Ler o lastRun do watermark (Lookup)

Adicione uma Lookup que leia watermark.json. No painel Settings, active First row only. Guarde o valor numa variável pipeline chamada lastWatermark usando um Set Variable com a expressão:

@activity('LookupWatermark').output.value.lastRun

Passo 5: Usar Get Metadata para obter lastModified dos ficheiros

Get Metadata com dataset da pasta (Field list: Child Items). A saída dá os nomes; para cada item quer-se o lastModified — pode usar um Lookup por ficheiro ou, preferível, habilitar dataset parametrizado e usar uma actividade ForEach sobre os childItems para executar um Get Metadata individual que retorne lastModified.

// Dentro do ForEach (items: @activity('GetFolder').output.childItems)
// Get Metadata (param: fileName) -> field: lastModified
// Expor output: activity('GetFileMetadata').output.lastModified

Passo 6: Filtrar ficheiros novos e limpar dados simples

Dentro do ForEach, após obter lastModified, use uma actividade If Condition para testar se lastModified > lastWatermark. Se true, executar um Data Flow ou Copy com mapping e validações simples (ex.: eliminar linhas com campos obrigatórios vazios, normalizar datas).

// If Condition expression
@greater(formatDateTime(activity('GetFileMetadata').output.lastModified,'yyyy-MM-ddTHH:mm:ssZ'), variables('lastWatermark'))

// Exemplo de transformações simples num Mapping Data Flow:
// - Source: CSV
// - Derived Column: trim() e parseDate()
// - Filter: isNotNull(campo_chave)
// - Sink: Azure SQL (modo upsert com chave única)

Passo 7: Carregamento incremental para Azure SQL

Para evitar duplicados, use no Sink do Data Flow a opção de update/insert (upsert) baseada numa coluna chave (por exemplo, id ou combinação de campos). Alternativa: carregar para uma tabela staging e executar stored procedure para deduplicar com MERGE.

// Exemplo minimal de MERGE (Azure SQL) para deduplicar após carga staging
MERGE dbo.Target AS T
USING dbo.Staging AS S
ON T.Key = S.Key
WHEN MATCHED THEN UPDATE SET T.Col = S.Col
WHEN NOT MATCHED THEN INSERT (Key, Col) VALUES (S.Key, S.Col);

Passo 8: Atualizar o watermark após sucesso

No final do pipeline, após confirmação de sucesso do carregamento, escreva no watermark.json a data/hora máxima dos ficheiros processados (por exemplo, now()). Use uma actividade Web para chamar a REST API do Blob (ou uma actividade Copy com dataset JSON de saída) para sobrescrever o ficheiro.

// Exemplo expressão para novo watermark
@utcNow()  // ou máximo entre ficheiros processados

// Para escrever com Copy activity: fonte = uma pequena tabela/variable, sink = dataset watermark.json

Verificar o resultado

Valide: 1) o pipeline executou sem erros; 2) a tabela alvo contém apenas registos esperados; 3) watermark.json foi actualizado com a nova data; 4) re-executar o pipeline não reprocessa ficheiros já processados. Use Monitor do Azure Data Factory e consultas na Azure SQL para ver alterações.

Conclusão

Com este padrão consegue processar ficheiros CSV incrementalmente, validar e carregar para Azure SQL reduzindo custo e duplicação. Próximos passos: adicionar logging detalhado, gerir erros com Dead-letter (staging) e parametrizar para várias pastas. Dica: comece por testar com poucos ficheiros e verifique sempre o timezone das datas.