Budowanie opóźnionych potoków

Programowanie równoległe z Dask w Pythonie

James Fulton

Climate Informatics Researcher

Fragmenty danych

Diagram przedstawiający zbiór danych, który mieści się na dysku twardym, ale jest zbyt duży dla pamięci RAM.

Diagram przedstawiający zbiór danych podzielony na fragmenty. Cały zbiór jest zbyt duży dla RAM, ale poszczególne fragmenty mieszczą się w pamięci.

Zbiór danych piosenek Spotify

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

Zbiór danych piosenek Spotify

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

Analiza danych

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)

Analiza danych

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)

Analiza danych

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

Używanie metod obiektu opóźnionego

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-...')

Metody i właściwości obiektów opóźnionych zwracają nowe obiekty opóźnione

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

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

Używanie metod obiektu opóźnionego

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-...')

Metody są wykonywane dopiero po wywołaniu .compute()

print(max_length.compute())

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

Używanie metod obiektu opóźnionego

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 to lista obiektów opóźnionych

print(maximums)

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

Obliczanie list obiektów opóźnionych

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],)

Obliczanie list obiektów opóźnionych

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]

Opóźniać czy nie opóźniać

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)

Głębszy graf zadań

absolute_maximum.visualize()

Graf zadań opisujący kroki potrzebne do obliczenia absolutnego maksimum w plikach. Graf jest duży, ale prosty.

Czas na ćwiczenia!

Programowanie równoległe z Dask w Pythonie