Impulsione seu workflow de ciência de dados com Dask: guia completo

Quando Wes McKinney começou a escrever o pandas, ele tinha uma regra de ouro: para o pandas funcionar de forma ideal, a RAM da máquina deve ser de 5 a 10 vezes maior que o conjunto de dados em questão. Essa regra era fácil de seguir por volta de 2010, mas agora já é 2023.

Em 2020, os datasets do mundo real já tinham crescido a tamanhos que derrubavam facilmente notebooks e máquinas do dia a dia. Prevendo esse problema com antecedência, uma solução foi lançada em 2015.

Dask é uma biblioteca open source desenvolvida pelos criadores do Anaconda para enfrentar os desafios de computação escalável e eficiente em grandes volumes de dados que excedem a memória de uma única máquina.

Este tutorial traz uma introdução completa ao Dask e seus recursos essenciais, incluindo interfaces para DataFrames, Arrays e Bags.

Configurando o Dask

Como qualquer outra biblioteca, você pode instalar o Dask de três formas: Conda, Pip e pelo código-fonte.

Como este é um artigo introdutório sobre Dask, não vamos cobrir o último método de instalação, que é voltado para mantenedores.

Se você usa Anaconda, o Dask já vem na instalação padrão (o que mostra o quanto a biblioteca é popular). Se quiser reinstalar ou atualizar, use o comando install:

conda install dask

A alternativa via PIP é a seguinte:

pip install "dask[complete]"

Adicionar a extensão [complete] também instala as dependências necessárias do Dask, evitando a necessidade de instalar NumPy, Pandas e Tornado manualmente.

Você pode verificar se a instalação foi bem-sucedida conferindo a versão da biblioteca:

import dask

dask.__version__

Saída:

'2023.5.0'

A maior parte do seu trabalho com Dask vai se concentrar em três interfaces: Dask DataFrames, Arrays e Bags. Vamos importá-las junto com numpy e pandas para usar no restante do artigo:

import dask.array as da
import dask.bag as db
import dask.dataframe as dd
import numpy as np
import pandas as pd

Conceitos básicos do Dask

De forma geral, você pode pensar no Dask como um "wrapper" que amplia as capacidades de ferramentas tradicionais como pandas, NumPy e Spark para lidar com datasets maiores que a memória.

Quando nos deparamos com objetos grandes, como arrays (vetores) ou matrizes (dataframes) maiores que a memória, o Dask os divide em blocos, também chamados de partições.

Por exemplo, considere o array de 12 números aleatórios em NumPy e em Dask:

narr = np.random.rand(12)

narr

array([0.9261154 , 0.87774082, 0.87078873, 0.22309476, 0.24575174,
      0.04182393, 0.31476305, 0.04599283, 0.62354124, 0.97597454,
      0.23923457, 0.81201211])

darr = da.from_array(narr, chunks=3)
darr

A imagem acima mostra que o array do Dask contém quatro blocos, pois definimos chunks como 3. Por baixo dos panos, cada bloco é um array do NumPy em si.

Agora, vamos considerar um exemplo bem maior. Vamos criar dois arrays de 10k por 100k (1 bilhão de elementos) e fazer multiplicação elemento a elemento em ambas as bibliotecas, medindo o desempenho:

# Create the NumPy arrays
arr1 = np.random.rand(10_000, 100_000)
arr2 = np.random.rand(10_000, 100_000)

# Create the Dask arrays
darr1 = da.from_array(arr1, chunks=(1_000, 10_000))
darr2 = da.from_array(arr2, chunks=(1_000, 10_000))

%%time

result_np = np.multiply(arr1, arr2)

Wall time: 3.19 s

%%time

result_dask = da.multiply(darr1, darr2)

Wall time: 94.8 ms

O código acima demonstra a multiplicação elemento a elemento de dois arrays grandes usando NumPy e Dask. Como mostrado na saída, o Dask é aproximadamente 34 vezes mais rápido que o NumPy para esse cálculo. Os ganhos de performance ficam ainda mais significativos à medida que a computação e o tamanho dos arrays aumentam.

O Dask usa uma abordagem semelhante de particionar e distribuir esses blocos entre todos os núcleos disponíveis da sua máquina para outros tipos de objeto também.

Dask DataFrames

Para aproveitar totalmente os benefícios do Dask, precisamos de um dataset grande, de preferência acima de 1 GB. No entanto, baixar um dataset assim para acompanhar o tutorial pode não ser o ideal. Em vez disso, você pode usar este script, que gera um dataset sintético com 10 milhões de linhas, 10 features numéricas e 10 categóricas.

Certifique-se de que sua máquina tenha pelo menos 12 GB de RAM para executar o script.

Quando o arquivo large_dataset.csv estiver no seu workspace, você pode carregá-lo usando a função read_csv da interface de Dask DataFrames (dd):

import dask.dataframe as dd

dask_df = dd.read_csv("data/large_dataset.csv")

dask_df.head()

Mesmo com um arquivo grande, você vai notar que o resultado é retornado quase instantaneamente. Para arquivos ainda maiores, você pode especificar o parâmetro `blocksize`, que define o número de bytes em que o arquivo será fragmentado.

Assim como Dask Arrays contêm blocos de pequenos arrays NumPy, o Dask é projetado para lidar com vários pequenos DataFrames do Pandas organizados ao longo do índice de linhas.

Como você pode perceber pela função read_csv, grande parte da sintaxe e das funcionalidades mais usadas da API do pandas é preservada no Dask. Os blocos de código a seguir devem soar familiares para quem já trabalhou com pandas.

Selecionando colunas e operações elemento a elemento

Neste exemplo, fazemos operações de coluna bem diretas no nosso Dask DataFrame, chamado dask_df. Somamos os valores da coluna Numeric_0 ao resultado da multiplicação dos valores de Numeric_9 e Numeric_3. Armazenamos o resultado em uma variável chamada result.

result = (
   dask_df["Numeric_0"] + dask_df["Numeric_9"] * dask_df["Numeric_3"]
)

result.compute().head()


0    1.301960
1    1.190679
2    1.100955
3    0.758272
4    0.926729
dtype: float64

Como comentamos, o Dask é um pouco diferente das ferramentas tradicionais porque não executa essas operações imediatamente. Em vez disso, ele cria um tipo de "plano" chamado task graph para realizar essas operações depois. Essa abordagem permite que o Dask otimize e paralelize os cálculos quando necessário. A função compute() dispara a execução e o head() apenas mostra as primeiras linhas do resultado.

Filtragem condicional

Agora, vamos ver como o Dask pode filtrar dados. Estamos selecionando as linhas do nosso DataFrame em que o valor na coluna "Categorical_5" é "A".

Esse processo de filtragem é semelhante ao do pandas, mas com um detalhe — o Dask faz a operação de forma preguiçosa. Ele prepara o task graph, mas espera para executar até chamarmos compute(). Quando rodamos head(), vemos as primeiras linhas do DataFrame filtrado.

dask_df[dask_df["Categorical_5"] == "A"].compute().head()

Estatísticas descritivas comuns

Em seguida, vamos gerar algumas estatísticas descritivas comuns usando a função describe() do Dask.

Ela fornece várias estatísticas do DataFrame, incluindo média, desvio padrão, mínimo, máximo e assim por diante. Como nos exemplos anteriores, o Dask prepara o task graph quando chamamos describe(), mas só executa quando chamamos compute().

dask_df.describe().compute()

dask_df["Categorical_3"].value_counts().compute().head()

Categorical_3
O    386038
C    385804
A    385493
P    385490
K    385116
Name: count, dtype: int64

Também usamos value_counts() para contar o número de ocorrências de cada valor único na coluna "Categorical_3". Disparamos a operação com compute(), e head() mostra os valores mais comuns.

Groupby

Por fim, vamos usar a função groupby() para agrupar nossos dados com base nos valores da coluna "Categorical_8". Em seguida, selecionamos a coluna "Numeric_7" e calculamos a média para cada grupo.

Isso é parecido com o uso de groupby() no pandas e, como você já deve imaginar, o Dask faz isso de forma preguiçosa. Disparamos a operação com compute(), e head() exibe a média da coluna "Numeric_7" para os primeiros grupos.

dask_df.groupby("Categorical_8")["Numeric_7"].mean().compute().head()


Categorical_8
A    0.498497
B    0.499767
C    0.500622
D    0.500307
E    0.499530
Name: Numeric_7, dtype: float64

Confira esta seção do guia do usuário do Dask para ver outras semelhanças entre pandas e Dask.

Avaliação preguiçosa (lazy evaluation)

Agora, vamos explorar o uso da função compute no fim de cada bloco de código.

O Dask avalia blocos de código em modo preguiçoso, em contraste com o modo ansioso (eager) do Pandas, que retorna resultados imediatamente.

Fazendo um paralelo com a cozinha, avaliação preguiçosa é como preparar os ingredientes e picar os legumes com antecedência, mas só juntar tudo para cozinhar quando necessário. A função compute cumpre esse papel.

Já a avaliação ansiosa é como jogar os ingredientes no fogo assim que ficam prontos. Essa abordagem faz tudo ficar pronto para servir de uma vez.

A avaliação preguiçosa é fundamental para o ótimo desempenho do Dask porque oferece:

1. Menos computação. As expressões só são avaliadas quando necessário (quando compute é chamado), evitando resultados intermediários desnecessários que podem nem ser usados no resultado final.
2. Alocação ideal de recursos. A avaliação preguiçosa evita alocar memória ou processamento para resultados intermediários que podem não ser necessários.
3. Suporte a grandes datasets. Esse método processa dados sob demanda ou em blocos menores, aproveitando melhor os recursos de memória.

Quando o resultado de compute é retornado, ele vem como Series/DataFrames do Pandas ou arrays do NumPy, e não como DataFrames nativos do Dask.

>>> type(dask_df)
dask.dataframe.core.DataFrame


>>> type(
   dask_df[["Numeric_5", "Numeric_6", "Numeric_7"]].mean().compute()
)

pandas.core.series.Series

O motivo é que a maioria das operações de manipulação de dados retorna apenas um subconjunto do dataframe original, ocupando muito menos espaço. Assim, não há necessidade de usar o paralelismo do Dask, e você pode continuar o restante do fluxo de trabalho no pandas ou NumPy.

Dask Bags e Dask Delayed para dados não estruturados

Dask Bags e Dask Delayed são dois componentes da biblioteca Dask que oferecem ferramentas poderosas para trabalhar com dados não estruturados ou semiestruturados, além de permitir avaliação preguiçosa.

Se antes dados tabulares eram os mais comuns, hoje os datasets frequentemente envolvem arquivos não estruturados como imagens, textos, vídeos e áudios. O Dask Bags fornece a funcionalidade e a API para lidar com esses arquivos de forma paralela e escalável.

Por exemplo, veja uma ilustração simples:

import dask.bag as db

# Create a Dask Bag from a list of strings
b = db.from_sequence(["apple", "banana", "orange", "grape", "kiwi"])

# Filter the strings that start with the letter 'a'
filtered_strings = b.filter(lambda x: x.startswith("a"))

# Map a function to convert each string to uppercase
uppercase_strings = filtered_strings.map(lambda x: x.upper())

# Compute the result as a list
result = uppercase_strings.compute()

print(result)
['APPLE']

Neste exemplo, criamos um Dask Bag b a partir de uma lista de strings. Em seguida, aplicamos operações no Bag para filtrar as strings que começam com a letra "a" e convertê-las para maiúsculas usando as funções filter() e map(), respectivamente. Por fim, calculamos o resultado como uma lista com compute() e imprimimos a saída.

Agora imagine executar operações bem mais complexas em bilhões de strings semelhantes armazenadas em um arquivo de texto. Sem a avaliação preguiçosa e o paralelismo do Dask Bags, isso seria um enorme desafio. (Leia mais sobre Bags na documentação do Dask).

Quanto ao Dask Delayed, ele oferece ainda mais flexibilidade e leva avaliação preguiçosa e paralelismo para vários outros cenários. Com o Dask Delayed, você pode converter qualquer função nativa do Python em um objeto preguiçoso usando o decorador @dask.delayed.

Aqui vai um exemplo simples:

%%time

import time
import dask


@dask.delayed
def process_data(x):
   # Simulate some computation
   time.sleep(1)
   return x**2


# Generate a list of inputs
inputs = range(1000)

# Apply the delayed function to each input
results = [process_data(x) for x in inputs]

# Compute the results in parallel
computed_results = dask.compute(*results)



Wall time: 42.1 s

Neste exemplo, definimos a função process_data decorada com @dask.delayed. Ela simula algum trabalho computacional dormindo 1 segundo e depois retorna o quadrado do valor de entrada.

Sem paralelismo, fazer esse cálculo para 1000 entradas levaria mais de 1000 segundos. Com Dask Delayed e execução paralela, a computação levou cerca de 42,1 segundos.

Esse exemplo mostra o poder do paralelismo em reduzir o tempo de processamento ao distribuir a carga de trabalho de forma eficiente entre vários núcleos ou workers.

É disso que se trata o paralelismo.

Conclusão e próximos passos

O Dask é uma das bibliotecas fundamentais do ecossistema de dados. Ele amplia a funcionalidade de bibliotecas queridinhas como NumPy, Pandas e Spark, permitindo lidar sem fricção com datasets maiores que a memória.

Com Dask Bags e Dask Delayed, ele leva paralelismo e avaliação preguiçosa para cenários menos tradicionais, como trabalho com dados não estruturados ou com objetos nativos do Python.

Para mergulhar nos detalhes, vale a pena ler com atenção a documentação do Dask.

Se você busca um recurso completo para dominar o Dask, confira nosso curso Parallel Programming With Dask in Python.