Dask के साथ अपने डेटा साइंस वर्कफ़्लो को बेहतर बनाएँ: एक व्यापक मार्गदर्शिका

जब Wes McKinney ने pandas लिखना शुरू किया, तो उनके पास एक सामान्य नियम था: pandas के इष्टतम रूप से काम करने के लिए, मशीन की RAM उस डेटासेट से 5–10 गुना बड़ी होनी चाहिए। यह नियम 2010 के आसपास मानना आसान था, लेकिन अब 2023 है।

2020 तक, वास्तविक दुनिया के डेटासेट इतने बड़े हो चुके थे कि वे रोज़मर्रा के लैपटॉप और मशीनों को आसानी से क्रैश कर सकते थे। इस समस्या का पहले से अनुमान लगाते हुए, 2015 में एक समाधान जारी किया गया।

Dask एक ओपन-सोर्स लाइब्रेरी है जिसे Anaconda के निर्माताओं ने विकसित किया है, ताकि एकल मशीन की मेमोरी क्षमता से बड़े डेटासेट्स पर स्केलेबल और कुशल कंप्यूटिंग की चुनौतियों से निपटा जा सके।

यह ट्यूटोरियल Dask और इसकी महत्वपूर्ण विशेषताओं का समग्र परिचय देता है, जिनमें DataFrames, Arrays और Bags के लिए इंटरफ़ेस शामिल हैं।

Dask सेट करना

अन्य किसी भी लाइब्रेरी की तरह, आप Dask को तीन तरीकों से इंस्टॉल कर सकते हैं: Conda, Pip, और सोर्स से।

चूंकि यह Dask पर एक प्रारंभिक लेख है, हम अंतिम इंस्टॉलेशन विधि को कवर नहीं करेंगे, क्योंकि वह मेंटेनर्स के लिए होती है।

यदि आप Anaconda का उपयोग करते हैं, तो Dask डिफ़ॉल्ट इंस्टॉलेशन में शामिल है (जो लाइब्रेरी की लोकप्रियता का संकेत है)। यदि आप इसे पुनः इंस्टॉल या अपग्रेड करना चाहें, तो आप install कमांड का उपयोग कर सकते हैं:

conda install dask

PIP विकल्प निम्नलिखित है:

pip install "dask[complete]"

[complete] एक्सटेंशन जोड़ने से Dask की आवश्यक dependencies भी इंस्टॉल हो जाती हैं, जिससे आपको NumPy, Pandas और Tornado को अलग से इंस्टॉल करने की ज़रूरत नहीं रहती।

आप लाइब्रेरी वर्ज़न देखकर जाँच सकते हैं कि इंस्टॉलेशन सफल रहा या नहीं:

import dask

dask.__version__

आउटपुट:

'2023.5.0'

आपका अधिकांश Dask कार्य तीन इंटरफ़ेस पर केंद्रित होगा: Dask DataFrames, Arrays और Bags। आइए इन्हें numpy और pandas के साथ इम्पोर्ट करें ताकि लेख के बाकी हिस्से में इनका उपयोग कर सकें:

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

Dask की मूल अवधारणाएँ

उच्च-स्तर पर, आप Dask को एक ऐसे रैपर के रूप में सोच सकते हैं जो pandas, NumPy और Spark जैसे पारंपरिक टूल्स की क्षमताओं का विस्तार करता है ताकि मेमोरी से बड़े डेटासेट्स को संभाला जा सके।

जब मेमोरी से बड़े ऐसे ऑब्जेक्ट्स सामने आते हैं जैसे एरेज़ (वेक्टर) या मैट्रिसेज़ (डेटाफ्रेम), तो Dask उन्हें छोटे टुकड़ों में बाँट देता है, जिन्हें पार्टिशन भी कहा जाता है।

उदाहरण के लिए, NumPy और Dask दोनों में 12 रैंडम संख्याओं के एरे पर विचार करें:

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

ऊपर की छवि दिखाती है कि Dask एरे में चार चंक्स हैं क्योंकि हमने chunks को 3 सेट किया है। अंदरूनी स्तर पर, प्रत्येक चंक स्वयं में एक NumPy एरे होता है।

अब, एक और बड़ा उदाहरण लेते हैं। हम 10k × 100k (1 बिलियन एलिमेंट्स) के दो एरे बनाएँगे और दोनों लाइब्रेरीज़ में एलिमेंट-वाइज़ गुणा करेंगे, साथ ही परफ़ॉर्मेंस भी मापेंगे:

# 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

ऊपर दिया गया कोड NumPy और Dask दोनों का उपयोग करके दो बड़े एरेज़ का एलिमेंट-वाइज़ गुणन दर्शाता है। जैसा कि आउटपुट में दिखाया गया है, इस गणना के लिए Dask लगभग 34 गुना तेज़ है। जैसे-जैसे गणना और एरे का आकार बढ़ता है, परफ़ॉर्मेंस लाभ और भी महत्वपूर्ण हो जाते हैं।

Dask अन्य ऑब्जेक्ट्स के लिए भी इसी तरह चंकिंग और इन चंक्स को आपकी मशीन के उपलब्ध सभी कोर्स में वितरित करने का तरीका अपनाता है।

Dask DataFrames

Dask के फ़ायदों को पूरी तरह समझने के लिए हमें एक बड़ा डेटासेट चाहिए, अधिमानतः 1 GB से अधिक। हालांकि, ट्यूटोरियल के साथ चलते हुए इतना बड़ा डेटासेट डाउनलोड करना व्यावहारिक नहीं होगा। इसकी बजाय, आप इस स्क्रिप्ट का उपयोग कर सकते हैं, जो 1 करोड़ पंक्तियों, 10 संख्यात्मक फीचर्स और 10 श्रेणीबद्ध फीचर्स वाला एक सिंथेटिक डेटासेट बनाती है।

कृपया सुनिश्चित करें कि आपकी मशीन में स्क्रिप्ट चलाने के लिए कम से कम 12 GB RAM हो।

एक बार जब आपके वर्कस्पेस में large_dataset.csv फ़ाइल आ जाए, तो आप इसे Dask DataFrames इंटरफ़ेस (dd) की read_csv फ़ंक्शन से लोड कर सकते हैं:

import dask.dataframe as dd

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

dask_df.head()

भले ही फ़ाइल बड़ी है, आप देखेंगे कि परिणाम लगभग तुरंत आ जाता है। और भी बड़ी फ़ाइलों के लिए, आप `blocksize` पैरामीटर निर्दिष्ट कर सकते हैं, जो यह निर्धारित करता है कि फ़ाइल को कितने बाइट्स के हिस्सों में बाँटना है।

जिस तरह Dask Arrays में छोटे NumPy एरेज़ के चंक्स होते हैं, उसी तरह Dask को पंक्ति सूचकांक के साथ व्यवस्थित अनेक छोटे Pandas DataFrames सँभालने के लिए डिज़ाइन किया गया है।

read_csv फ़ंक्शन से आप समझ ही गए होंगे कि pandas API की अधिकांश सामान्य रूप से प्रयुक्त सिंटैक्स और कार्यक्षमता Dask में संरक्षित है। निम्नलिखित कोड ब्लॉक्स आपको pandas के साथ काम करने के अपने अनुभव से परिचित लगेंगे।

कॉलम चुनना और एलिमेंट-वाइज़ ऑपरेशंस

इस उदाहरण में, हम अपने Dask DataFrame, dask_df, पर कुछ सीधे-सरल कॉलम ऑपरेशंस कर रहे हैं। हम Numeric_0 कॉलम के मानों को Numeric_9 और Numeric_3 के गुणनफल में जोड़ रहे हैं। परिणाम को 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

जैसा कि हमने बताया, Dask पारंपरिक टूल्स से थोड़ा अलग है क्योंकि यह इन ऑपरेशंस को तुरंत निष्पादित नहीं करता। इसके बजाय, यह इन्हें बाद में करने के लिए एक तरह की 'योजना' बनाता है जिसे टास्क ग्राफ कहा जाता है। यह तरीका Dask को गणनाओं का अनुकूलन करने और आवश्यकता पड़ने पर उन्हें समानांतर (parallel) करने देता है। compute() फ़ंक्शन Dask को अंततः इन गणनाओं को निष्पादित करने के लिए ट्रिगर करता है, और head() हमें परिणाम की शुरुआती कुछ पंक्तियाँ दिखाता है।

शर्त आधारित फ़िल्टरिंग

अब देखते हैं कि Dask डेटा को फ़िल्टर कैसे कर सकता है। हम अपने DataFrame से वे पंक्तियाँ चुन रहे हैं जहाँ "Categorical_5" कॉलम का मान "A" है।

यह फ़िल्टरिंग प्रक्रिया pandas जैसी ही है, बस एक अंतर है—Dask इसे लेज़ी तरीके से करता है। यह इस ऑपरेशन के लिए टास्क ग्राफ तैयार करता है लेकिन तब तक निष्पादित होने का इंतज़ार करता है जब तक हम compute() नहीं बुलाते। जब हम head() चलाते हैं, तो हमें फ़िल्टर किए गए DataFrame की शुरुआती कुछ पंक्तियाँ मिलती हैं।

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

सामान्य सार सांख्यिकी

अगले चरण में, हम Dask के describe() फ़ंक्शन का उपयोग करके कुछ सामान्य सार सांख्यिकी जनरेट करेंगे।

यह हमारे DataFrame के लिए औसत, मानक विचलन, न्यूनतम, अधिकतम आदि जैसी वर्णनात्मक सांख्यिकी देता है। पिछले उदाहरणों की तरह, Dask describe() कॉल करते समय इस ऑपरेशन के लिए टास्क ग्राफ तैयार करता है, लेकिन इसका निष्पादन तब तक रोकता है जब तक हम 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

हम value_counts() का उपयोग "Categorical_3" कॉलम में प्रत्येक अद्वितीय मान की आवृत्ति गिनने के लिए भी करते हैं। compute() के साथ ऑपरेशन ट्रिगर होता है, और head() हमें सबसे सामान्य मान दिखाता है।

Groupby

अंत में, groupby() फ़ंक्शन का उपयोग करके "Categorical_8" कॉलम के मानों के आधार पर अपने डेटा को समूहित करें। फिर "Numeric_7" कॉलम चुनें और प्रत्येक समूह के लिए औसत निकालें।

यह pandas में 'groupby()' के उपयोग जैसा ही है, लेकिन जैसा कि आपने अनुमान लगाया होगा, Dask इसे लेज़ी तरीके से करता है। हम compute() के साथ ऑपरेशन ट्रिगर करते हैं, और head() शुरुआती कुछ समूहों के लिए "Numeric_7" कॉलम का औसत दिखाता है।

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 और Dask के बीच बाकी समानताओं पर Dask यूज़र गाइड के इस अनुभाग पर नज़र डालें।

लेज़ी इवैल्यूएशन

अब, आइए प्रत्येक कोड ब्लॉक के अंत में compute फ़ंक्शन के उपयोग का अन्वेषण करें।

Pandas के ईगर मोड, जो तुरंत परिणाम देता है, की तुलना में Dask कोड ब्लॉक्स का मूल्यांकन लेज़ी मोड में करता है।

कुकिंग के संदर्भ में तुलना करें, तो लेज़ी इवैल्यूएशन ऐसा है जैसे आप सामग्री तैयार कर लें और सब्ज़ियाँ काट लें, लेकिन उन्हें केवल ज़रूरत पड़ने पर पकाने के लिए मिलाएँ। compute फ़ंक्शन वही काम करता है।

इसके विपरीत, ईगर इवैल्यूएशन ऐसा है जैसे सामग्री तैयार होते ही उन्हें पका देना। यह तरीका सुनिश्चित करता है कि सब कुछ एक साथ परोसने के लिए तैयार हो।

लेज़ी इवैल्यूएशन Dask के उत्कृष्ट परफ़ॉर्मेंस की कुंजी है क्योंकि यह प्रदान करता है:

1. घटी हुई गणना। अभिव्यक्तियों का मूल्यांकन केवल तब होता है जब उनकी ज़रूरत हो (जब compute बुलाया जाए), जिससे उन मध्यवर्ती परिणामों से बचा जा सके जो अंतिम परिणाम में प्रयुक्त नहीं होने वाले हैं।
2. संसाधनों का इष्टतम आवंटन। लेज़ी इवैल्यूएशन उन मध्यवर्ती परिणामों के लिए मेमोरी या प्रोसेसिंग पावर आवंटित करने से बचता है जिनकी आवश्यकता नहीं पड़ सकती।
3. बड़े डेटासेट्स के लिए समर्थन। यह तरीका डेटा एलिमेंट्स को ऑन-द-फ्लाई या छोटे चंक्स में प्रोसेस करता है, जिससे मेमोरी संसाधनों का कुशल उपयोग संभव होता है।

compute के परिणाम लौटने पर, उन्हें नेटिव Dask DataFrames के बजाय Pandas Series/DataFrames या NumPy एरेज़ के रूप में दिया जाता है।

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


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

pandas.core.series.Series

इसका कारण यह है कि अधिकांश डेटा हेर-फेर (manipulation) ऑपरेशंस केवल मूल डेटाफ्रेम के एक उपसमुच्चय को लौटाते हैं, जो बहुत कम स्थान घेरता है। इसलिए, Dask के पैरेललिज़्म का उपयोग करने की आवश्यकता नहीं रहती, और आप अपने वर्कफ़्लो के बाकी हिस्से pandas या NumPy में जारी रख सकते हैं।

असंरचित डेटा के लिए Dask Bags और Dask Delayed

Dask Bags और Dask Delayed, Dask लाइब्रेरी के दो घटक हैं जो असंरचित या अर्द्ध-संरचित डेटा के साथ काम करने और लेज़ी इवैल्यूएशन सक्षम करने के लिए शक्तिशाली उपकरण प्रदान करते हैं।

जहाँ पहले तालिका-बद्ध डेटा सबसे आम था, वहीं आज के डेटासेट्स में अक्सर छवियाँ, टेक्स्ट फ़ाइलें, वीडियो और ऑडियो जैसी असंरचित फ़ाइलें शामिल होती हैं। Dask Bags ऐसी असंरचित फ़ाइलों को समानांतर और स्केलेबल तरीके से संभालने की कार्यक्षमता और API प्रदान करता है।

उदाहरण के लिए, एक सरल चित्रण पर विचार करें:

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

इस उदाहरण में, हम स्ट्रिंग्स की एक सूची से Dask Bag b बनाते हैं। फिर हम Bag पर ऑपरेशंस लागू करते हैं ताकि 'a' अक्षर से शुरू होने वाली स्ट्रिंग्स को फ़िल्टर किया जा सके और filter() और map() फ़ंक्शंस के माध्यम से उन्हें अपरकेस में बदला जा सके। अंत में, हम compute() मेथड का उपयोग करके परिणाम को एक सूची के रूप में निकालते हैं और आउटपुट प्रिंट करते हैं।

अब कल्पना करें कि आप किसी टेक्स्ट फ़ाइल में संग्रहीत अरबों समान स्ट्रिंग्स पर इससे भी अधिक जटिल ऑपरेशंस कर सकते हैं। Dask Bags द्वारा प्रदान की गई लेज़ी इवैल्यूएशन और पैरेललिज़्म के बिना, आपको भारी चुनौतियों का सामना करना पड़ता। (Bags के बारे में अधिक पढ़ें Dask डॉक्यूमेंटेशन में)।

जहाँ तक Dask Delayed की बात है, यह और भी अधिक लचीलापन प्रदान करता है और विभिन्न अन्य परिदृश्यों में लेज़ी इवैल्यूएशन और पैरेललिज़्म लाता है। Dask Delayed के साथ, आप @dask.delayed डेकोरेटर का उपयोग करके किसी भी नेटिव Python फ़ंक्शन को लेज़ी ऑब्जेक्ट में बदल सकते हैं।

यहाँ एक सरल उदाहरण है:

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

इस उदाहरण में, हम @dask.delayed से सज्जित process_data फ़ंक्शन परिभाषित करते हैं। यह फ़ंक्शन 1 सेकंड के लिए स्लीप करके कुछ कम्प्यूटेशनल कार्य का सिमुलेशन करता है और फिर इनपुट मान का वर्ग लौटा देता है।

पैरेललिज़्म के बिना, 1000 इनपुट्स पर यह गणना करने में 1000 सेकंड से अधिक समय लगता। लेकिन Dask Delayed और समानांतर निष्पादन के साथ, यह गणना केवल लगभग 42.1 सेकंड में पूरी हो गई।

यह उदाहरण कार्यभार को कई कोर्स या वर्कर्स में कुशलतापूर्वक बाँटकर गणना समय घटाने में पैरेललिज़्म की शक्ति को दर्शाता है।

यही तो पैरेललिज़्म का सार है।

निष्कर्ष और आगे के संसाधन

Dask डेटा इकोसिस्टम की आधारभूत लाइब्रेरीज़ में से एक है। यह NumPy, Pandas और Spark जैसी प्रिय लाइब्रेरीज़ की कार्यक्षमता का विस्तार करता है, जिससे मेमोरी से बड़े डेटासेट्स को सहजता से संभालना संभव होता है।

Dask Bags और Dask Delayed के साथ, यह असंरचित डेटा या नेटिव Python ऑब्जेक्ट्स जैसे अलहदा परिदृश्यों में भी पैरेललिज़्म और लेज़ी इवैल्यूएशन लाता है।

सूक्ष्म विवरणों के लिए, Dask डॉक्यूमेंटेशन ज़रूर पढ़ें।

यदि आप Dask में महारत हासिल करने के लिए एक व्यापक संसाधन खोज रहे हैं, तो हमारा कोर्स देखें  Parallel Programming With Dask in Python।