विलंबित पाइपलाइनों का निर्माण

Python में Dask के साथ Parallel Programming

James Fulton

Climate Informatics Researcher

डेटा के चंक्स

एक डायग्राम जो दिखाता है कि डेटासेट हार्ड ड्राइव में तो समा सकता है, लेकिन RAM में नहीं।

एक डायग्राम जो दिखाता है कि डेटासेट को कई हिस्सों में बाँटा गया है। पूरा डेटासेट RAM में नहीं समाएगा, पर प्रत्येक हिस्सा समा जाएगा।

Spotify songs डेटासेट

files = [
  '2005_tracks.csv',
  '2006_tracks.csv',
  '2007_tracks.csv',
  '2008_tracks.csv',
  '2009_tracks.csv',
  '2010_tracks.csv',
  ...
  '2020_tracks.csv',
]

Spotify songs डेटासेट

                       name  duration_ms release_date  ...
0     Aldrig (feat. Carmon)       247869   2019-01-01  ...
2  2019 - The Year to Build       288105   2019-01-01  ...
3                 Na zawsze       186812   2019-01-01  ...
4         Humo en la Trampa       258354   2019-01-01  ...
5                     Au Au       176000   2019-01-01  ...
...                     ...          ...          ...  ...

डेटा का विश्लेषण

import pandas as pd

maximums = []

for file in files:
    # Load each file
    df = pd.read_csv(file)

    # Find maximum track length in each file
    max_length = df['duration_ms'].max()

    # Store this maximum
    maximums.append(max_length)


# Find the maximum of all the maximum lengths
absolute_maximum = max(maximums)

डेटा का विश्लेषण

import pandas as pd

maximums = []

for file in files:
    # Load each file
    df = delayed(pd.read_csv)(file) # <------- delay loading
    # Find maximum track length in each file
    max_length = df['duration_ms'].max()
    # Store this maximum
    maximums.append(max_length)

# Find the maximum of all the maximum lengths
absolute_maximum = max(maximums)

डेटा का विश्लेषण

import pandas as pd

maximums = []

for file in files:
    # Load each file
    df = delayed(pd.read_csv)(file) # <------- delay loading
    # Find maximum track length in each file
    max_length = df['duration_ms'].max()
    # Store this maximum
    maximums.append(max_length)

# Find the maximum of all the maximum lengths
absolute_maximum = delayed(max)(maximums) # <------- delay max() function

Delayed ऑब्जेक्ट की methods का उपयोग

import pandas as pd

maximums = []

for file in files:
    df = delayed(pd.read_csv)(file)
    # Use the .max() method
    max_length = df['duration_ms'].max()

    maximums.append(max_length)


absolute_maximum = delayed(max)(maximums)

print(max_length)

Delayed('max-0602855d-3ee6-4c43-a4d2-...')

Delayed ऑब्जेक्ट की methods और properties नए delayed ऑब्जेक्ट लौटाती हैं

print(df.shape)
print(df.shape.compute())

Delayed('getattr-bc1e8838ab...')
(11907, 12)

Delayed ऑब्जेक्ट की methods का उपयोग

import pandas as pd

maximums = []

for file in files:
    df = delayed(pd.read_csv)(file)
    # Use a method which doesn't exist
    max_length = df['duration_ms'].fake()

    maximums.append(max_length)


absolute_maximum = delayed(max)(maximums)

print(max_length)

Delayed('max-6c026036-5daf-4b2-...')

Methods तब तक नहीं चलतीं जब तक .compute() न बुलाएँ

print(max_length.compute())

...
AttributeError: 'Series' object has no 
attribute 'fake'

Delayed ऑब्जेक्ट की methods का उपयोग

import pandas as pd

maximums = []

for file in files:
    df = delayed(pd.read_csv)(file)

    max_length = df['duration_ms'].max()
    # Add delayed object to list
    maximums.append(max_length)

# Run delayed max on delayed objects list
absolute_maximum = delayed(max)(maximums)

maximums delayed ऑब्जेक्ट्स की एक लिस्ट है

print(maximums)

[Delayed('max-80b...'), 
Delayed('max-fa15d...', 
...]

Delayed ऑब्जेक्ट्स की सूची compute करना

import pandas as pd

maximums = []

for file in files:
    df = delayed(pd.read_csv)(file)

    max_length = df['duration_ms'].max()
    # Add dalayed object to list
    maximums.append(max_length)

# Compute all the maximums
all_maximums = dask.compute(maximums)

print(all_maximums)

([2539418, 4368000, ...
... 4511716, 4864333],)

Delayed ऑब्जेक्ट्स की सूची compute करना

import pandas as pd

maximums = []

for file in files:
    df = delayed(pd.read_csv)(file) 

    max_length = df['duration_ms'].max()

    maximums.append(max_length)

# Compute all the maximums
all_maximums = dask.compute(maximums)[0]

print(all_maximums)

[2539418, 4368000, ...
... 4511716, 4864333]

Delay करना है या नहीं

def get_max_track(df):
    return df['duration_ms'].max()

for file in files:
    df = delayed(pd.read_csv)(file) 
    # Use function to find max
    max_length = get_max_track(df)

    maximums.append(max_length)


absolute_maximum = delayed(max)(maximums)

और गहरा टास्क ग्राफ

absolute_maximum.visualize()

एक टास्क ग्राफ जो फाइलों में absolute maximum निकालने के स्टेप्स दिखाता है। ग्राफ बड़ा है पर सरल है।

अभ्यास करते हैं!

Python में Dask के साथ Parallel Programming