Dask ile Veri Bilimi İş Akışınızı Hızlandırın: Kapsamlı Bir Rehber

Wes McKinney pandas'ı yazmaya başladığında şu kuralı benimsemişti: pandas'ın en iyi şekilde çalışması için makinenin RAM boyutu, söz konusu veri setinden 5–10 kat daha büyük olmalıdır. Bu kurala 2010 civarında uymak kolaydı, ancak artık yıl 2023.

2020'ye gelindiğinde, gerçek dünyadaki veri setleri çoktan sıradan dizüstü bilgisayarları ve makineleri kolayca çökertme boyutlarına ulaşmıştı. Bu sorun önceden öngörülerek, 2015'te bir çözüm yayımlandı.

Dask, tek bir makinenin bellek kapasitesini aşan büyük veri setlerinde ölçeklenebilir ve verimli hesaplama zorluklarını çözmek için Anaconda'nın yaratıcıları tarafından geliştirilmiş, açık kaynaklı bir kütüphanedir.

Bu eğitim, Dask'a ve DataFrame, Array ve Bag arayüzleri de dahil olmak üzere kritik özelliklerine kapsamlı bir giriş sunar.

Dask'ı Kurma

Diğer tüm kütüphaneler gibi Dask'ı üç şekilde kurabilirsiniz: Conda, Pip ve kaynaktan.

Bu, Dask hakkında giriş niteliğinde bir yazı olduğundan, bakım yapanlara yönelik olan son kurulum yöntemini ele almayacağız.

Anaconda kullanıyorsanız, Dask varsayılan kurulumunuza dahildir (bu, kütüphanenin ne kadar popüler olduğunun da bir göstergesidir). Yeniden kurmak veya yükseltmek isterseniz install komutunu kullanabilirsiniz:

conda install dask

PIP alternatifi ise şöyledir:

pip install "dask[complete]"

[complete] ekini kullanmak, Dask'ın gerekli bağımlılıklarını da kurar ve NumPy, Pandas ve Tornado'yu elle kurma ihtiyacını ortadan kaldırır.

Kurulumun başarılı olup olmadığını kütüphane sürümüne bakarak kontrol edebilirsiniz:

import dask

dask.__version__

Çıktı:

'2023.5.0'

Dask ile yapacağınız işlerin çoğu üç arayüze odaklanır: Dask DataFrame, Array ve Bag. Yazının geri kalanında kullanmak üzere bunları numpy ve pandas ile birlikte içe aktaralım:

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

Dask'ın Temel Kavramları

Yüksek seviyede, Dask'ı, pandas, NumPy ve Spark gibi geleneksel araçların yeteneklerini belleği aşan veri kümelerini ele alacak şekilde genişleten bir sarmalayıcı olarak düşünebilirsiniz.

Belleği aşan diziler (vektörler) veya matrisler (dataframe'ler) gibi büyük nesnelerle karşılaşıldığında Dask bunları, bölüm olarak da adlandırılan parçalara ayırır.

Örneğin, hem NumPy hem de Dask'ta 12 rastgele sayıdan oluşan diziyi ele alalım:

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

Yukarıdaki görselde, chunks değerini 3 olarak ayarladığımız için Dask dizisinin dört parçadan oluştuğu görülüyor. Arka planda, her bir parça başlı başına bir NumPy dizisidir.

Şimdi çok daha büyük bir örnek düşünelim. İki adet 10k x 100k dizi (1 milyar eleman) oluşturacağız ve her iki kütüphanede de eleman düzeyinde çarpma yaparken performansı ölçeceğiz:

# 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

Yukarıdaki kod, iki büyük dizinin NumPy ve Dask ile eleman düzeyinde çarpımını göstermektedir. Çıktıdan görüldüğü üzere, bu hesaplamada Dask, NumPy'den yaklaşık 34 kat daha hızlıdır. Hesaplama ve dizi boyutu arttıkça performans kazanımları daha da belirginleşir.

Dask, diğer nesneler için de benzer bir parçalara ayırma yaklaşımı kullanır ve bu parçaları makinenizdeki tüm kullanılabilir çekirdeklere dağıtır.

Dask DataFrame'leri

Dask'ın faydalarını tam olarak görmek için tercihen 1 GB'ın üzerinde büyük bir veri setine ihtiyaç vardır. Ancak, eğitimi takip etmek için böyle bir veri setini indirmek her zaman en iyi seçenek olmayabilir. Bunun yerine, 10 milyon satır, 10 sayısal özellik ve 10 kategorik özellik içeren sentetik bir veri seti üreten bu betiği kullanabilirsiniz.

Lütfen betiği çalıştırmak için makinenizde en az 12 GB RAM bulunduğundan emin olun.

large_dataset.csv dosyası çalışma alanınızda olduğunda, Dask DataFrame arayüzünün (dd) read_csv işleviyle içeri alabilirsiniz:

import dask.dataframe as dd

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

dask_df.head()

Dosya büyük olmasına rağmen sonucun neredeyse anında geldiğini fark edeceksiniz. Daha da büyük dosyalar için, dosyanın kaç bayta bölüneceğini belirleyen `blocksize` parametresini belirtebilirsiniz.

Dask Array'lerin küçük NumPy dizilerinden oluşan parçaları içerdiğine benzer şekilde, Dask satır indeksine göre sıralanmış birden çok küçük Pandas DataFrame'ini ele alacak şekilde tasarlanmıştır.

read_csv işlevinden de tahmin edebileceğiniz gibi, pandas API'sinin yaygın olarak kullanılan sözdizimi ve işlevselliğinin çoğu Dask'ta korunmuştur. Aşağıdaki kod blokları, pandas ile çalışırken size tanıdık gelecektir.

Sütun seçimi ve eleman düzeyinde işlemler

Bu örnekte, dask_df adlı Dask DataFrame'imiz üzerinde oldukça basit sütun işlemleri yapıyoruz. Numeric_0 sütunundaki değerleri, Numeric_9 ile Numeric_3 sütunlarının çarpımına ekliyoruz. Sonucu result adlı bir değişkende saklıyoruz.

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

Belirttiğimiz gibi, Dask geleneksel araçlardan biraz farklıdır; bu işlemleri hemen çalıştırmak yerine, daha sonra gerçekleştirmek üzere bir tür "plan" yani görev grafiği oluşturur. Bu yaklaşım Dask'ın hesaplamaları optimize etmesini ve gerektiğinde paralelleştirmesini sağlar. compute() işlevi, Dask'ın bu hesaplamaları nihayet gerçekleştirmesini tetikler ve head() sadece sonucun ilk birkaç satırını gösterir.

Koşullu filtreleme

Şimdi, Dask'ın verileri nasıl filtreleyebildiğine bakalım. DataFrame'imizde "Categorical_5" sütunundaki değerin "A" olduğu satırları seçiyoruz.

Bu filtreleme işlemi pandas'ta yaptığınıza benzer, ancak bir farkla — Dask bu işlemi tembel şekilde yapar. Bu işlem için görev grafiğini hazırlar ancak compute() çağrılana kadar yürütmeyi bekler. head() çalıştırıldığında ise, filtrelenmiş DataFrame'imizin ilk birkaç satırını görürüz.

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

Yaygın özet istatistikler

Sırada Dask'ın describe() işlevini kullanarak bazı yaygın özet istatistikleri üretmek var.

Bu işlev DataFrame'imiz için ortalama, standart sapma, minimum, maksimum vb. bir dizi betimsel istatistik verir. Önceki örneklerimizde olduğu gibi, Dask describe() çağrıldığında bu işlem için görev grafiğini hazırlar ancak compute() çağrılana kadar yürütmeyi bekler.

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

Ayrıca "Categorical_3" sütunundaki her bir benzersiz değerin kaç kez göründüğünü saymak için value_counts() kullanıyoruz. İşlemi compute() ile tetikliyor ve head() en yaygın değerleri gösteriyor.

Groupby

Son olarak, verilerimizi "Categorical_8" sütunundaki değerlere göre gruplamak için groupby() işlevini kullanalım. Ardından "Numeric_7" sütununu seçip her grup için ortalamasını hesaplıyoruz.

Bu, pandas'ta 'groupby()' kullanımınıza benzer, ama tahmin edeceğiniz üzere Dask bunu tembel şekilde yapar. İşlemi compute() ile tetikleriz ve head() ilk birkaç grup için "Numeric_7" sütununun ortalamasını gösterir.

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

pandas ile Dask arasındaki diğer benzerlikler için Dask kullanıcı kılavuzunun bu bölümüne göz atın.

Tembel değerlendirme

Şimdi, her kod bloğunun sonunda compute işlevinin kullanımını inceleyelim.

Dask, sonuçları anında döndüren Pandas'ın hevesli (eager) moduna kıyasla kod bloklarını tembel (lazy) modda değerlendirir.

Yemek pişirmeye benzetirsek, tembel değerlendirme, malzemeleri önceden hazırlayıp sebzeleri doğramaya ama pişirmek için yalnızca gerektiğinde birleştirmeye benzer. compute işlevi bu amaca hizmet eder.

Buna karşılık, hevesli değerlendirme, malzemeler hazır olur olmaz ateşe atıp pişirmeye benzer. Bu yaklaşım her şeyin aynı anda servise hazır olmasını sağlar.

Tembel değerlendirme, Dask'ın üstün performansının anahtarıdır; çünkü şunları sağlar:

1. Azaltılmış hesaplama. İfadeler yalnızca ihtiyaç duyulduğunda ( compute çağrıldığında) değerlendirilir; böylece nihai sonuçta kullanılmayabilecek gereksiz ara sonuçlardan kaçınılır.
2. En uygun kaynak tahsisi. Tembel değerlendirme, gerekli olmayabilecek ara sonuçlara bellek veya işlem gücü ayırmaktan kaçınır.
3. Büyük veri setleri için destek. Bu yöntem, veri öğelerini anında veya daha küçük parçalara bölerek işler ve bellek kaynaklarının verimli kullanımını sağlar.

compute sonuçları döndürüldüğünde, yerel Dask DataFrame'leri yerine Pandas Series/DataFrame'leri veya NumPy dizileri olarak verilir.

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


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

pandas.core.series.Series

Bunun nedeni, çoğu veri işleme işleminin, orijinal dataframe'in yalnızca bir alt kümesini döndürmesi ve çok daha az yer kaplamasıdır. Dolayısıyla Dask'ın paralelliğini kullanmaya gerek kalmaz; iş akışınızın geri kalanına pandas veya NumPy ile devam edebilirsiniz.

Yapılandırılmamış Veri için Dask Bag ve Dask Delayed

Dask Bag ve Dask Delayed, yapılandırılmamış veya yarı yapılandırılmış verilerle çalışmak ve tembel değerlendirmeyi etkinleştirmek için güçlü araçlar sunan Dask kütüphanesinin iki bileşenidir.

Geçmişte tabular veri en yaygın türken, günümüzdeki veri setleri sıklıkla görseller, metin dosyaları, videolar ve sesler gibi yapılandırılmamış dosyaları içerir. Dask Bag, bu tür yapılandırılmamış dosyaları paralel ve ölçeklenebilir bir şekilde ele alacak işlevsellik ve API sağlar.

Örneğin, basit bir örnek düşünelim:

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']

Bu örnekte, bir dizi string'den bir Dask Bag b oluşturuyoruz. Ardından filter() ve map() işlevlerini kullanarak 'a' harfiyle başlayan string'leri filtreliyor ve bunları büyük harfe dönüştürüyoruz. Son olarak compute() yöntemiyle sonucu bir liste olarak hesaplayıp çıktıyı yazdırıyoruz.

Şimdi, bir metin dosyasında saklanan benzer milyarlarca string üzerinde çok daha karmaşık işlemler yapabildiğinizi hayal edin. Dask Bag'in sunduğu tembel değerlendirme ve paralellik olmadan ciddi zorluklarla karşılaşırdınız. (Bag hakkında daha fazla bilgi için Dask dokümantasyonuna bakın.).

Dask Delayed ise daha da fazla esneklik sunar ve tembel değerlendirme ile paralelliği çeşitli diğer senaryolara taşır. Dask Delayed ile herhangi bir yerel Python işlevini @dask.delayed dekoratörüyle tembel bir nesneye dönüştürebilirsiniz.

Basit bir örnek:

%%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

Bu örnekte, @dask.delayed ile süslenmiş bir process_data işlevi tanımlıyoruz. İşlev, 1 saniye uyuyarak bazı hesaplamaları simüle eder ve ardından giriş değerinin karesini döndürür.

Paralellik olmadan, bu hesaplamayı 1000 girdi üzerinde yapmak 1000 saniyeden uzun sürerdi. Ancak Dask Delayed ve paralel yürütme ile hesaplama yalnızca yaklaşık 42,1 saniye sürdü.

Bu örnek, iş yükünü birden çok çekirdek veya işçi üzerine verimli şekilde dağıtarak hesaplama süresini azaltmadaki paralelliğin gücünü gösterir.

Paralelliğin özü budur.

Sonuç ve Ek Kaynaklar

Dask, veri ekosisteminin temel kütüphanelerinden biridir. NumPy, Pandas ve Spark gibi sevilen kütüphanelerin işlevselliğini genişleterek belleği aşan veri kümelerini sorunsuzca ele almanızı sağlar.

Dask Bag ve Dask Delayed ile, yapılandırılmamış verilerle çalışma veya yerel Python nesneleriyle çalışma gibi alışılmadık senaryolara paralellik ve tembel değerlendirmeyi taşır.

Ayrıntılar için Dask dokümanlarını mutlaka baştan sona okuyun.

Dask'ı kapsamlı şekilde öğrenebileceğiniz bir kaynak arıyorsanız, kursumuza göz atın: Parallel Programming With Dask in Python.