dask

O que dask faz?

Visão Geral

Dask é uma biblioteca Python para computação paralela e distribuída que oferece três capacidades críticas:

• Execução maior que a memória em máquinas únicas para dados que excedem a RAM disponível
• Processamento paralelo para melhorar a velocidade computacional em múltiplos núcleos
• Computação distribuída suportando datasets em escala de terabytes em várias máquinas

Dask escala de laptops (processando ~100 GiB) a clusters (processando ~100 TiB) mantendo APIs Python familiares.

Versão upstream atual: dask 2026.3.0 (PyPI, março de 2026). Documentação: docs.dask.org. Desde a 2025.1.0, a API DataFrame baseada em expressões com planejamento de consultas é a única implementação — não instale dask-expr separadamente nem defina dataframe.query-planning: False.

Início Rápido

Instalação

uv pip install "dask>=2025.1"

Para um workflow típico de pandas/NumPy com o agendador distribuído e o painel de monitoramento:

uv pip install "dask[complete]"

Armazenamento remoto de objetos (S3, GCS, Azure):

uv pip install s3fs    # caminhos s3://
uv pip install gcsfs   # caminhos gs://

Requer Python 3.10+ (suporte a 3.9 removido em 2024.12). I/O de DataFrames requer PyArrow 16+ (a partir de dask 2026.1.2).

Quando Usar Esta Habilidade

Esta habilidade deve ser usada quando:

• Processar datasets que excedem a RAM disponível
• Escalar operações pandas ou NumPy para datasets maiores
• Paralelizar computações para melhorias de desempenho
• Processar múltiplos arquivos eficientemente (CSVs, Parquet, JSON, logs de texto)
• Construir workflows paralelos personalizados com dependências de tarefas
• Distribuir cargas de trabalho entre múltiplos núcleos ou máquinas

Capacidades Principais

Dask oferece cinco componentes principais, cada um adequado a diferentes casos de uso:

1. DataFrames - Operações Pandas Paralelas

Propósito: Escalar operações pandas para datasets maiores através de processamento paralelo.

Quando Usar:

• Dados tabulares excedem a RAM disponível
• Necessidade de processar múltiplos arquivos CSV/Parquet juntos
• Operações pandas estão lentas e precisam de paralelização
• Escalar de protótipo pandas para produção

Documentação de Referência: Para orientação abrangente sobre DataFrames Dask, consulte references/dataframes.md que inclui:

• Leitura de dados (arquivo único, múltiplos arquivos, padrões glob)
• Operações comuns (filtragem, groupby, joins, agregações)
• Operações personalizadas com map_partitions
• Dicas de otimização de desempenho
• Padrões comuns (ETL, séries temporais, processamento multiarquivo)

Exemplo Rápido:

import dask.dataframe as dd

# Ler múltiplos arquivos como um único DataFrame
ddf = dd.read_csv('dados/2024-*.csv')

# Operações são lentas até compute()
filtrado = ddf[ddf['valor'] > 100]
resultado = filtrado.groupby('categoria').mean().compute()

Pontos-Chave:

• Operações são lentas (constroem grafo de tarefas) até .compute() ser chamado
• Use map_partitions para operações personalizadas eficientes
• Converta para DataFrame cedo ao trabalhar com dados estruturados de outras fontes

2. Arrays - Operações NumPy Paralelas

Propósito: Estender as capacidades do NumPy para datasets maiores que a memória usando algoritmos em blocos.

Quando Usar:

• Arrays excedem a RAM disponível
• Operações NumPy precisam de paralelização
• Trabalhar com datasets científicos (HDF5, Zarr, NetCDF)
• Necessidade de álgebra linear paralela ou operações com arrays

Documentação de Referência: Para orientação abrangente sobre Arrays Dask, consulte references/arrays.md que inclui:

• Criação de arrays (a partir de NumPy, aleatório, a partir de disco)
• Estratégias de chunk e otimização
• Operações comuns (aritmética, reduções, álgebra linear)
• Operações personalizadas com map_blocks
• Integração com HDF5, Zarr e XArray

Exemplo Rápido:

import dask.array as da

# Criar array grande com chunks
x = da.random.random((100000, 100000), chunks=(10000, 10000))

# Operações são lentas
y = x + 100
z = y.mean(axis=0)

# Computar resultado
resultado = z.compute()

Pontos-Chave:

• O tamanho do chunk é crítico (busque ~100 MB por chunk)
• Operações trabalham em chunks em paralelo
• Rechunk os dados quando necessário para operações eficientes
• Use map_blocks para operações não disponíveis no Dask

3. Bags - Processamento Paralelo de Dados Não Estruturados

Propósito: Processar dados não estruturados ou semiestruturados (texto, JSON, logs) com operações funcionais.

Quando Usar:

• Processar arquivos de texto, logs ou registros JSON
• Limpeza de dados e ETL antes da análise estruturada
• Trabalhar com objetos Python que não se encaixam em formatos de array/dataframe
• Necessidade de processamento eficiente em streaming com baixo uso de memória

Documentação de Referência: Para orientação abrangente sobre Bags Dask, consulte references/bags.md que inclui:

• Leitura de arquivos de texto e JSON
• Operações funcionais (map, filter, fold, groupby)
• Conversão para DataFrames
• Padrões comuns (análise de logs, processamento JSON, processamento de texto)
• Considerações de desempenho

Exemplo Rápido:

import dask.bag as db
import json

# Ler e analisar arquivos JSON
bag = db.read_text('logs/*.json').map(json.loads)

# Filtrar e transformar
válido = bag.filter(lambda x: x['status'] == 'válido')
processado = válido.map(lambda x: {'id': x['id'], 'valor': x['valor']})

# Converter para DataFrame para análise
ddf = processado.to_dataframe()

Pontos-Chave:

• Use para limpeza inicial de dados, depois converta para DataFrame/Array
• Use foldby em vez de groupby para melhor desempenho
• Operações são streaming e eficientes em memória
• Converta para formatos estruturados (DataFrame) para operações complexas

4. Futures - Paralelização Baseada em Tarefas

Propósito: Construir workflows paralelos personalizados com controle refinado sobre execução de tarefas e dependências.

Quando Usar:

• Construir workflows dinâmicos e evolutivos
• Necessidade de execução imediata de tarefas (não lenta)
• Computações dependem de condições em tempo de execução
• Implementar algoritmos paralelos personalizados
• Necessidade de computações com estado

Documentação de Referência: Para orientação abrangente sobre Futures Dask, consulte references/futures.md que inclui:

• Configuração de cliente distribuído
• Submissão de tarefas e trabalho com futures
• Dependências de tarefas e movimento de dados
• Coordenação avançada (filas, locks, eventos, atores)
• Padrões comuns (varredura de parâmetros, tarefas dinâmicas, algoritmos iterativos)

Exemplo Rápido:

from dask.distributed import Client

client = Client()  # Criar cluster local

# Submeter tarefas (executa imediatamente)
def processar(x):
    return x ** 2

futures = client.map(processar, range(100))

# Reunir resultados
resultados = client.gather(futures)

client.close()

Pontos-Chave:

• Requer cliente distribuído (mesmo para máquina única)
• Tarefas executam imediatamente quando submetidas
• Pré-dispersar dados grandes para evitar transferências repetidas
• ~1ms de overhead por tarefa (não adequado para milhões de tarefas pequenas)
• Use atores para workflows com estado

5. Agendadores - Backends de Execução

Propósito: Controlar como e onde as tarefas Dask executam (threads, processos, distribuído).

Quando Escolher Agendador:

• Threads (padrão): Operações NumPy/Pandas, bibliotecas que liberam GIL, benefício de memória compartilhada
• Processos: Código Python puro, processamento de texto, operações com GIL
• Síncrono: Depuração com pdb, profiling, entendimento de erros
• Distribuído: Necessidade de painel de monitoramento, clusters multimáquina, funcionalidades avançadas

Documentação de Referência: Para orientação abrangente sobre Agendadores Dask, consulte references/schedulers.md que inclui:

• Descrições detalhadas dos agendadores e suas características
• Métodos de configuração (global, gerenciador de contexto, por computação)
• Considerações de desempenho e overhead
• Padrões comuns e solução de problemas
• Configuração de threads para desempenho ideal

Exemplo Rápido:

import dask
import dask.dataframe as dd

# Usar threads para DataFrame (padrão, bom para numérico)
ddf = dd.read_csv('dados.csv')
resultado1 = ddf.mean().compute()  # Usa threads

# Usar processos para trabalho pesado em Python
import dask.bag as db
bag = db.read_text('logs/*.txt')
resultado2 = bag.map(função_python).compute(scheduler='processes')

# Usar síncrono para depuração
dask.config.set(scheduler='synchronous')
resultado3 = computação_problemática.compute()  # Pode usar pdb

# Usar distribuído para monitoramento e escalabilidade
from dask.distributed import Client
client = Client()
resultado4 = computação.compute()  # Usa distribuído com painel

Pontos-Chave:

• Threads: Menor overhead (~10 µs/tarefa), melhor para trabalho numérico
• Processos: Evita GIL (~10 ms/tarefa), melhor para trabalho Python
• Distribuído: Painel de monitoramento (~1 ms/tarefa), escala para clusters
• Pode trocar de agendador por computação ou globalmente

Melhores Práticas

Para orientação abrangente sobre otimização de desempenho, estratégias de gerenciamento de memória e armadilhas comuns a evitar, consulte references/best-practices.md. Princípios-chave incluem:

Comece com Soluções Mais Simples

Antes de usar Dask, explore:

• Melhores algoritmos
• Formatos de arquivo eficientes (Parquet em vez de CSV)
• Código compilado (Numba, Cython)
• Amostragem de dados

Regras Críticas de Desempenho

1. Não Carregue Dados Localmente e Depois Entregue ao Dask

# Errado: Carrega todos os dados na memória primeiro
import pandas as pd
df = pd.read_csv('grande.csv')
ddf = dd.from_pandas(df, npartitions=10)

# Correto: Deixe o Dask lidar com o carregamento
import dask.dataframe as dd
ddf = dd.read_csv('grande.csv')

2. Evite Chamadas Repetidas a compute()

# Errado: Cada compute é separado
for item in itens:
    resultado = computação_dask(item).compute()

# Correto: Compute único para todos
computações = [computação_dask(item) for item in itens]
resultados = dask.compute(*computações)

3. Não Construa Grafos de Tarefas Excessivamente Grandes

• Aumente tamanhos de chunk se houver milhões de tarefas
• Use map_partitions/map_blocks para fundir operações
• Verifique o tamanho do grafo de tarefas: len(ddf.__dask_graph__())

4. Escolha Tamanhos de Chunk Adequados

• Alvo: ~100 MB por chunk (ou 10 chunks por núcleo na memória do worker)
• Muito grande: Estouro de memória
• Muito pequeno: Overhead de agendamento

5. Use o Painel de Monitoramento

from dask.distributed import Client
client = Client()
print(client.dashboard_link)  # Monitorar desempenho, identificar gargalos

Padrões Comuns de Workflow

Pipeline ETL

import dask.dataframe as dd

# Extrair: Ler dados
ddf = dd.read_csv('dados_brutos/*.csv')

# Transformar: Limpar e processar
ddf = ddf[ddf['status'] == 'válido']
ddf['quantidade'] = ddf['quantidade'].astype('float64')
ddf = ddf.dropna(subset=['coluna_importante'])

# Carregar: Agregar e salvar
resumo = ddf.groupby('categoria').agg({'quantidade': ['sum', 'mean']})
resumo.to_parquet('saida/resumo.parquet')

Pipeline de Não Estruturado para Estruturado

import dask.bag as db
import json

# Começar com Bag para dados não estruturados
bag = db.read_text('logs/*.json').map(json.loads)
bag = bag.filter(lambda x: x['status'] == 'válido')

# Converter para DataFrame para análise estruturada
ddf = bag.to_dataframe()
resultado = ddf.groupby('categoria').mean().compute()

Computação de Array em Grande Escala

import dask.array as da

# Carregar ou criar array grande
x = da.from_zarr('grande_dataset.zarr')

# Processar em chunks
normalizado = (x - x.mean()) / x.std()

# Salvar resultado (use mode= para sobrescrever; zarr_array_kwargs para compressão)
da.to_zarr(normalizado, 'normalizado.zarr', mode='w')

Workflow Paralelo Personalizado

from dask.distributed import Client

client = Client()

# Dispersar dataset grande uma vez
dados = client.scatter(grande_dataset)

# Processar em paralelo com dependências
futures = []
for param in parâmetros:
    future = client.submit(processar, dados, param)
    futures.append(future)

# Reunir resultados
resultados = client.gather(futures)

Selecionando o Componente Certo

Use este guia de decisão para escolher o componente Dask apropriado:

Tipo de Dado:

• Dados tabulares → DataFrames
• Arrays numéricos → Arrays
• Texto/JSON/logs → Bags (depois converter para DataFrame)
• Objetos Python personalizados → Bags ou Futures

Tipo de Operação:

• Operações pandas padrão → DataFrames
• Operações NumPy padrão → Arrays
• Tarefas paralelas personalizadas → Futures
• Processamento de texto/ETL → Bags

Nível de Controle:

• Alto nível, automático → DataFrames/Arrays
• Baixo nível, manual → Futures

Tipo de Workflow:

• Grafo de computação estático → DataFrames/Arrays/Bags
• Dinâmico, evolutivo → Futures

Considerações de Integração

Formatos de Arquivo

• Eficientes: Parquet, HDF5, Zarr (colunar, comprimido, amigável para paralelismo)
• Compatíveis, mas mais lentos: CSV (use apenas para ingestão inicial)
• Para Arrays: HDF5, Zarr, NetCDF

Conversão entre Coleções

# Bag → DataFrame
ddf = bag.to_dataframe()

# DataFrame → Array (para dados numéricos)
arr = ddf.to_dask_array(lengths=True)

# Array → DataFrame
ddf = dd.from_dask_array(arr, columns=['col1', 'col2'])

Com Outras Bibliotecas

• XArray: Envolve arrays Dask com dimensões rotuladas (geoespacial, imagens)
• Dask-ML: Aprendizado de máquina com APIs compatíveis com scikit-learn
• Distributed: Gerenciamento avançado de clusters e monitoramento

Depuração e Desenvolvimento

Workflow de Desenvolvimento Iterativo

1. Teste em dados pequenos com agendador síncrono:

dask.config.set(scheduler='synchronous')
resultado = computação.compute()  # Pode usar pdb, depuração fácil

2. Valide com threads em amostra:

amostra = ddf.head(1000)  # Amostra pequena
# Teste a lógica, depois escale para dataset completo

3. Escalone com distribuído para monitoramento:

from dask.distributed import Client
client = Client()
print(client.dashboard_link)  # Monitorar desempenho
resultado = computação.compute()

Problemas Comuns

Erros de Memória:

• Diminua tamanhos de chunk
• Use persist() estrategicamente e exclua quando terminar
• Verifique vazamentos de memória em funções personalizadas

Início Lento:

• Grafo de tarefas muito grande (aumente tamanhos de chunk)
• Use map_partitions ou map_blocks para reduzir tarefas

Paralelização Ruim:

• Chunks muito grandes (aumente o número de partições)
• Uso de threads com código Python (troque para processos)
• Dependências de dados impedem paralelismo

Ferramentas de Depuração

• Agendador Síncrono: dask.config.set(scheduler='synchronous') para depuração passo a passo
• Função visualize(): ddf.visualize() mostra o grafo de computação
• Painel Distribuído: Métricas em tempo real, gráficos de progresso, perfil de tarefas
• Amostragem de Dados: Teste com ddf.head(1000) antes da execução completa

Exemplos de Referência

Consulte os arquivos de referência para exemplos detalhados de cada componente:

• references/dataframes.md - Operações abrangentes com DataFrames
• references/arrays.md - Operações abrangentes com Arrays
• references/bags.md - Operações abrangentes com Bags
• references/futures.md - Operações abrangentes com Futures
• references/schedulers.md - Configuração de agendadores
• references/best-practices.md - Estratégias de otimização de desempenho