Dask

Dask

Visão Geral

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

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

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

Quando Usar Esta Skill

Esta skill 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 performance
• Processar múltiplos arquivos eficientemente (CSVs, Parquet, JSON, logs de texto)
• Construir workflows paralelos customizados com dependências de tarefas
• Distribuir cargas de trabalho entre múltiplos núcleos ou máquinas

Capacidades Principais

Dask fornece cinco componentes principais, cada um adequado para diferentes casos de uso:

1. DataFrames - Operações Paralelas com Pandas

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 são lentas e precisam de paralelização
• Escalar de prototipagem pandas para produção

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

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

Exemplo Rápido:

import dask.dataframe as dd

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

# Operações são lazy até compute() ser chamado
filtered = ddf[ddf['value'] > 100]
result = filtered.groupby('category').mean().compute()

Pontos-Chave:

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

2. Arrays - Operações Paralelas com NumPy

Propósito: Estender capacidades NumPy para datasets maiores que a memória usando algoritmos de 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 completa sobre Arrays Dask, consulte references/arrays.md que inclui:

• Criar arrays (de NumPy, aleatório, do disco)
• Estratégias de chunking e otimização
• Operações comuns (aritmética, reduções, álgebra linear)
• Operações customizadas 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 lazy
y = x + 100
z = y.mean(axis=0)

# Computar resultado
result = z.compute()

Pontos-Chave:

• Tamanho de chunk é crítico (buscar ~100 MB por chunk)
• Operações funcionam em chunks em paralelo
• Rechunk dados quando necessário para operações eficientes
• Use map_blocks para operações não disponíveis em Dask

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

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

Quando Usar:

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

Documentação de Referência: Para orientação completa 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 performance

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
valid = bag.filter(lambda x: x['status'] == 'valid')
processed = valid.map(lambda x: {'id': x['id'], 'value': x['value']})

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

Pontos-Chave:

• Use para limpeza inicial de dados, depois converta para DataFrame/Array
• Use foldby em vez de groupby para melhor performance
• 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 customizados com controle fino 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 lazy)
• Computações dependem de condições de runtime
• Implementar algoritmos paralelos customizados
• Necessidade de computações com estado

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

• Configurar cliente distribuído
• Submeter tarefas e trabalhar com futures
• Dependências de tarefas e movimento de dados
• Coordenação avançada (filas, locks, eventos, actors)
• 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 process(x):
    return x ** 2

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

# Coletar resultados
results = client.gather(futures)

client.close()

Pontos-Chave:

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

5. Schedulers - Backends de Execução

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

Quando Escolher Scheduler:

• Threads (padrão): Operações NumPy/Pandas, bibliotecas que liberam GIL, benefício de memória compartilhada
• Processes: Código Python puro, processamento de texto, operações ligadas ao GIL
• Synchronous: Debugging com pdb, profiling, entender erros
• Distributed: Necessidade de dashboard, clusters multi-máquina, recursos avançados

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

• Descrições detalhadas de schedulers e características
• Métodos de configuração (global, context manager, por-compute)
• Considerações de performance e overhead
• Padrões comuns e troubleshooting
• Configuração de threads para performance ótima

Exemplo Rápido:

import dask
import dask.dataframe as dd

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

# Usar processes para trabalho pesado em Python
import dask.bag as db
bag = db.read_text('logs/*.txt')
result2 = bag.map(python_function).compute(scheduler='processes')

# Usar synchronous para debugging
dask.config.set(scheduler='synchronous')
result3 = problematic_computation.compute()  # Pode usar pdb

# Usar distributed para monitoramento e escala
from dask.distributed import Client
client = Client()
result4 = computation.compute()  # Usa distributed com dashboard

Pontos-Chave:

• Threads: Menor overhead (~10 µs/tarefa), melhor para trabalho numérico
• Processes: Evita GIL (~10 ms/tarefa), melhor para trabalho Python
• Distributed: Dashboard de monitoramento (~1 ms/tarefa), escala para clusters
• Pode trocar schedulers por computação ou globalmente

Melhores Práticas

Para orientação abrangente sobre otimização de performance, 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:

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

Regras Críticas de Performance

1. Não Carregue Dados Localmente Depois Passe para Dask

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

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

2. Evite Chamadas Repetidas a compute()

# Errado: Cada compute é separado
for item in items:
    result = dask_computation(item).compute()

# Correto: Um único compute para todos
computations = [dask_computation(item) for item in items]
results = dask.compute(*computations)

3. Não Construa Grafos de Tarefas Excessivamente Grandes

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

4. Escolha Tamanhos de Chunk Apropriados

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

5. Use o Dashboard

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

Padrões de Workflow Comuns

Pipeline ETL

import dask.dataframe as dd

# Extract: Ler dados
ddf = dd.read_csv('raw_data/*.csv')

# Transform: Limpar e processar
ddf = ddf[ddf['status'] == 'valid']
ddf['amount'] = ddf['amount'].astype('float64')
ddf = ddf.dropna(subset=['important_col'])

# Load: Agregar e salvar
summary = ddf.groupby('category').agg({'amount': ['sum', 'mean']})
summary.to_parquet('output/summary.parquet')

Pipeline 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'] == 'valid')

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

Computação com Array em Larga Escala

import dask.array as da

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

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

# Salvar resultado
da.to_zarr(normalized, 'normalized.zarr')

Workflow Paralelo Customizado

from dask.distributed import Client

client = Client()

# Scatter dataset grande uma única vez
data = client.scatter(large_dataset)

# Processar em paralelo com dependências
futures = []
for param in parameters:
    future = client.submit(process, data, param)
    futures.append(future)

# Coletar resultados
results = client.gather(futures)

Selecionando o Componente Certo

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

Tipo de Dados:

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

Tipo de Operação:

• Operações pandas padrão → DataFrames
• Operações NumPy padrão → Arrays
• Tarefas paralelas customizadas → 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 (colunares, comprimidos, amigos de paralelização)
• Compatíveis mas 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, imaging)
• Dask-ML: Machine learning com APIs compatíveis com scikit-learn
• Distributed: Gerenciamento avançado de cluster e monitoramento

Debugging e Desenvolvimento

Workflow de Desenvolvimento Iterativo

1. Testar em dados pequenos com scheduler synchronous:

dask.config.set(scheduler='synchronous')
result = computation.compute()  # Pode usar pdb, fácil debugging

2. Validar com threads em amostra:

sample = ddf.head(1000)  # Amostra pequena
# Testar lógica, depois escalar para dataset completo

3. Escalar com distributed para monitoramento:

from dask.distributed import Client
client = Client()
print(client.dashboard_link)  # Monitorar performance
result = computation.compute()

Problemas Comuns

Erros de Memória:

• Diminuir tamanhos de chunk
• Usar persist() estrategicamente e deletar quando pronto
• Verificar memory leaks em funções customizadas

Início Lento:

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

Paralelização Pobre:

• Chunks muito grandes (aumentar número de partições)
• Usar threads com código Python (trocar para processes)
• Dependências de dados impedindo paralelismo

Arquivos de Referência

Todos os arquivos de documentação de referência podem ser lidos conforme necessário para informações detalhadas:

• references/dataframes.md - Guia completo de DataFrame Dask
• references/arrays.md - Guia completo de Array Dask
• references/bags.md - Guia completo de Bag Dask
• references/futures.md - Guia completo de Futures Dask e computação distribuída
• references/schedulers.md - Guia completo de seleção e configuração de scheduler
• references/best-practices.md - Otimização abrangente de performance e troubleshooting

Carregar estes arquivos quando usuários precisarem de informações detalhadas sobre componentes específicos Dask, operações ou padrões além da orientação rápida fornecida aqui.