Je data science-workflow boosten met Dask: een uitgebreide gids

Toen Wes McKinney begon met het schrijven van pandas, hanteerde hij een vuistregel: om pandas optimaal te laten werken, moet de RAM-grootte van de machine 5–10 keer groter zijn dan de dataset in kwestie. Deze regel was rond 2010 makkelijk te volgen, maar inmiddels is het 2023.

Rond 2020 waren echte datasets al gegroeid tot formaten die alledaagse laptops en machines eenvoudig konden laten crashen. Met dit probleem in het vooruitzicht werd in 2015 een oplossing uitgebracht.

Dask is een open-sourcelibrary, ontwikkeld door de makers van Anaconda, om de uitdagingen van schaalbare en efficiënte computing op grote datasets die het geheugen van één machine overstijgen, aan te pakken.

Deze tutorial biedt een uitgebreide introductie tot Dask en de belangrijkste features, waaronder interfaces voor DataFrames, Arrays en Bags.

Dask installeren

Net als elke andere library kun je Dask op drie manieren installeren: Conda, Pip en vanuit de broncode.

Omdat dit een introductie-artikel over Dask is, behandelen we de laatste installatiemethode niet, aangezien die voor maintainers bedoeld is.

Als je Anaconda gebruikt, is Dask inbegrepen in je standaardinstallatie (wat aangeeft hoe populair de library is). Als je het opnieuw wilt installeren of upgraden, kun je het install-commando gebruiken:

conda install daskMogelijk gemaakt door

Het PIP-alternatief is het volgende:

pip install "dask[complete]"Mogelijk gemaakt door

Door de extensie [complete] toe te voegen, worden ook de vereiste dependencies van Dask geïnstalleerd, zodat je NumPy, Pandas en Tornado niet handmatig hoeft te installeren.

Je kunt controleren of de installatie is gelukt door naar de libraryversie te kijken:

import dask

dask.__version__Mogelijk gemaakt door

Output:

'2023.5.0'Mogelijk gemaakt door

Het meeste van je Dask-werk zal zich richten op drie interfaces: Dask DataFrames, Arrays en Bags. Laten we ze importeren, samen met numpy en pandas, voor de rest van het artikel:

import dask.array as da
import dask.bag as db
import dask.dataframe as dd
import numpy as np
import pandas as pdMogelijk gemaakt door

Basisconcepten van Dask

Op hoog niveau kun je Dask zien als een wrapper die de mogelijkheden van traditionele tools zoals pandas, NumPy en Spark uitbreidt om datasets te verwerken die groter zijn dan het geheugen.

Wanneer je te maken hebt met grote objecten, zoals arrays (vectoren) of matrices (dataframes) die groter zijn dan het geheugen, splitst Dask ze op in stukken, ook wel partities genoemd.

Neem bijvoorbeeld de array van 12 willekeurige getallen in zowel NumPy als Dask:

narr = np.random.rand(12)

narrMogelijk gemaakt door

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])Mogelijk gemaakt door

darr = da.from_array(narr, chunks=3)
darrMogelijk gemaakt door

De bovenstaande afbeelding laat zien dat de Dask-array vier chunks bevat, omdat we chunks op 3 hebben gezet. Onder de motorkap is elke chunk op zichzelf een NumPy-array.

Laten we nu een veel groter voorbeeld bekijken. We maken twee arrays van 10k bij 100k (1 miljard elementen) en voeren elementgewijze vermenigvuldiging uit in beide libraries, terwijl we de performance meten:

# 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))Mogelijk gemaakt door

%%time

result_np = np.multiply(arr1, arr2)Mogelijk gemaakt door

Wall time: 3.19 sMogelijk gemaakt door

%%time

result_dask = da.multiply(darr1, darr2)Mogelijk gemaakt door

Wall time: 94.8 msMogelijk gemaakt door

De bovenstaande code demonstreert de elementgewijze vermenigvuldiging van twee grote arrays met zowel NumPy als Dask. Zoals in de output te zien is, is Dask voor deze berekening ongeveer 34 keer sneller dan NumPy. De performancewinst wordt nog groter naarmate de berekening en arraygrootte toenemen.

Dask gebruikt een vergelijkbare aanpak van chunking en het distribueren van deze chunks over alle beschikbare cores op je machine voor andere objecten.

Dask DataFrames

Om de voordelen van Dask volledig te waarderen, hebben we een grote dataset nodig, bij voorkeur groter dan 1 GB. Maar zo'n dataset downloaden om de tutorial te volgen is misschien niet ideaal. In plaats daarvan kun je dit script gebruiken, dat een synthetische dataset genereert met 10 miljoen rijen, 10 numerieke features en 10 categorische features.

Zorg ervoor dat je machine minstens 12 GB RAM heeft om het script uit te voeren.

Zodra je het bestand large_dataset.csv in je werkmap hebt, kun je het inladen met de functie read_csv van de Dask DataFrames-interface (dd):

import dask.dataframe as dd

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

dask_df.head()Mogelijk gemaakt door

Hoewel het bestand groot is, zie je dat het resultaat vrijwel direct wordt opgehaald. Voor nog grotere bestanden kun je de parameter `blocksize` opgeven, die bepaalt in hoeveel bytes het bestand wordt opgedeeld.

Net zoals Dask Arrays uit chunks van kleine NumPy-arrays bestaan, is Dask ontworpen om meerdere kleine Pandas DataFrames te verwerken die langs de rij-index zijn gerangschikt.

Zoals je kunt afleiden uit de functie read_csv, is het grootste deel van de veelgebruikte syntax en functionaliteit van de pandas-API behouden in Dask. De volgende codeblokken zouden je bekend moeten voorkomen uit je werk met pandas.

Kolommen selecteren en elementgewijze operaties

In dit voorbeeld doen we vrij eenvoudige kolombewerkingen op onze Dask DataFrame met de naam dask_df. We tellen de waarden uit de kolom Numeric_0 op bij het resultaat van het vermenigvuldigen van de waarden uit Numeric_9 en Numeric_3. We slaan de uitkomst op in een variabele met de naam 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: float64Mogelijk gemaakt door

Zoals we al noemden, werkt Dask net wat anders dan traditionele tools doordat het deze operaties niet onmiddellijk uitvoert. In plaats daarvan maakt het een soort 'plan', een task graph, om deze operaties later uit te voeren. Deze aanpak stelt Dask in staat de berekeningen te optimaliseren en waar nodig te paralleliseren. De functie compute() triggert Dask om de berekeningen uiteindelijk uit te voeren, en head() laat ons de eerste paar rijen van het resultaat zien.

Voorwaardelijk filteren

Laten we nu kijken hoe Dask data kan filteren. We selecteren rijen uit onze DataFrame waar de waarde in de kolom "Categorical_5" gelijk is aan "A".

Dit filterproces lijkt op hoe je het in pandas zou doen, maar met een twist: Dask doet deze operatie lui. Het bereidt de task graph voor deze operatie voor, maar wacht met uitvoeren tot we compute() aanroepen. Als we head() draaien, zien we de eerste paar rijen van onze gefilterde DataFrame.

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

Veelvoorkomende samenvattende statistieken

Vervolgens gaan we een aantal veelgebruikte samenvattende statistieken genereren met de functie describe() van Dask.

Die geeft ons een handvol beschrijvende statistieken voor onze DataFrame, waaronder het gemiddelde, de standaarddeviatie, minimum, maximum, enzovoort. Net als in onze eerdere voorbeelden bereidt Dask de task graph voor deze operatie voor wanneer we describe() aanroepen, maar wacht het met uitvoeren tot we compute() aanroepen.

dask_df.describe().compute()Mogelijk gemaakt door

dask_df["Categorical_3"].value_counts().compute().head()Mogelijk gemaakt door

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

We gebruiken ook value_counts() om het aantal voorkomens te tellen van elke unieke waarde in de kolom "Categorical_3". We triggeren de operatie met compute(), en head() laat ons de meest voorkomende waarden zien.

Groupby

Tot slot gebruiken we de functie groupby() om onze data te groeperen op basis van waarden in de kolom "Categorical_8". Daarna selecteren we de kolom "Numeric_7" en berekenen we het gemiddelde voor elke groep.

Dit lijkt op hoe je groupby() in pandas zou gebruiken, maar zoals je al vermoedde doet Dask dit lui. We triggeren de operatie met compute(), en head() toont het gemiddelde van de kolom "Numeric_7" voor de eerste paar groepen.

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
Mogelijk gemaakt door

Bekijk deze sectie van de Dask-gebruikersgids voor de rest van de overeenkomsten tussen pandas en Dask.

Luie evaluatie

Laten we nu het gebruik van de functie compute aan het einde van elk codeblok verkennen.

Dask evalueert codeblokken in luie modus, in tegenstelling tot de gretige modus van Pandas, die direct resultaten teruggeeft.

Om een parallel te trekken met koken: luie evaluatie is alsof je ingrediënten alvast voorbereidt en groenten snijdt, maar ze pas samenvoegt om te koken wanneer dat nodig is. De functie compute vervult die rol.

Gretige evaluatie daarentegen is alsof je ingrediënten direct in de pan gooit zodra ze klaar zijn. Deze aanpak zorgt ervoor dat alles tegelijk serveerklaar is.

Luie evaluatie is cruciaal voor de uitstekende performance van Dask, omdat het het volgende biedt:

1. Minder berekeningen. Expressies worden alleen geëvalueerd wanneer dat nodig is (wanneer compute wordt aangeroepen), waardoor onnodige tussenresultaten die mogelijk niet in het eindresultaat worden gebruikt, worden vermeden.
2. Optimale resource-allocatie. Luie evaluatie voorkomt dat er geheugen of rekenkracht wordt toegewezen aan tussenresultaten die mogelijk niet nodig zijn.
3. Ondersteuning voor grote datasets. Deze methode verwerkt data-elementen on-the-fly of in kleinere chunks, waardoor het geheugen efficiënt wordt benut.

Wanneer de resultaten van compute worden teruggegeven, zijn het Pandas Series/DataFrames of NumPy-arrays in plaats van native Dask DataFrames.

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


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

pandas.core.series.Series
Mogelijk gemaakt door

De reden hiervoor is dat de meeste datamanipulaties slechts een subset van de oorspronkelijke dataframe teruggeven, die veel minder ruimte inneemt. Er is dan geen noodzaak om de parallelisatie van Dask te gebruiken, en je kunt de rest van je workflow vervolgen in pandas of NumPy.

Dask Bags en Dask Delayed voor ongestructureerde data

Dask Bags en Dask Delayed zijn twee componenten van de Dask-library die krachtige tools bieden voor het werken met ongestructureerde of semi-gestructureerde data en het mogelijk maken van luie evaluatie.

Waar in het verleden tabeldata het meest voorkwam, bevatten hedendaagse datasets vaak ongestructureerde bestanden zoals afbeeldingen, tekstbestanden, video en audio. Dask Bags biedt de functionaliteit en API om zulke ongestructureerde bestanden parallel en schaalbaar te verwerken.

Neem bijvoorbeeld de volgende eenvoudige illustratie:

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']Mogelijk gemaakt door

In dit voorbeeld maken we een Dask Bag b van een lijst met strings. We passen vervolgens operaties toe op de Bag om de strings te filteren die beginnen met de letter 'a' en ze om te zetten naar hoofdletters met de functies filter() en map(). Tot slot berekenen we het resultaat als een lijst met de methode compute() en printen we de output.

Stel je nu voor dat je nog complexere operaties kunt uitvoeren op miljarden vergelijkbare strings die in een tekstbestand zijn opgeslagen. Zonder de luie evaluatie en het parallelisme van Dask Bags zou je tegen aanzienlijke uitdagingen aanlopen. (Lees meer over Bags in de Dask-documentatie).

Wat betreft Dask Delayed: dit biedt nog meer flexibiliteit en introduceert luie evaluatie en parallelisme in verschillende andere scenario's. Met Dask Delayed kun je elke native Python-functie omzetten in een lui object met de decorator @dask.delayed.

Hier is een eenvoudig voorbeeld:

%%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
Mogelijk gemaakt door

In dit voorbeeld definiëren we een functie process_data met de decorator @dask.delayed. De functie simuleert wat rekenwerk door 1 seconde te slapen en geeft daarna het kwadraat van de invoerwaarde terug.

Zonder parallelisme zou deze berekening op 1000 inputs meer dan 1000 seconden hebben geduurd. Maar met Dask Delayed en parallelle uitvoering duurde de berekening slechts ongeveer 42,1 seconden.

Dit voorbeeld demonstreert de kracht van parallelisme bij het verkorten van rekentijd door de werkbelasting efficiënt te verdelen over meerdere cores of workers.

Daar draait parallelisme om.

Conclusie en verdere bronnen

Dask is een van de hoeksteenlibraries in het data-ecosysteem. Het breidt de functionaliteit uit van geliefde libraries als NumPy, Pandas en Spark, waardoor je naadloos kunt werken met datasets die groter zijn dan het geheugen.

Met Dask Bags en Dask Delayed brengt het parallelisme en luie evaluatie naar onconventionele scenario's, zoals werken met ongestructureerde data of met native Python-objecten.

Voor de fijne kneepjes kun je de Dask-docs uitgebreid doornemen.

Ben je op zoek naar een complete bron om Dask te masteren? Bekijk dan onze cursus Parallel Programming With Dask in Python.