<jupyter_start><jupyter_text>CSYE7245 - Lab | Dask https://dask.org/<jupyter_code>import dask.dataframe as dd
import pandas as pd
from tqdm import tqdm_notebook
from dask.distributed import Client
import glob
import os
import numpy as np
import math
%%time
taxi_data = dd.read_csv('tripdata.csv')<jupyter_output>CPU times: user 16.2 ms, sys: 3.82 ms, total: 20 ms
Wall time: 22.3 ms<jupyter_text>Unlike `pandas`, you just get the structure of the dataframe and NOT its contents here.<jupyter_code>taxi_data<jupyter_output><empty_output><jupyter_text>This is because Dask has loaded the dataframe in several chunks, which are present on the disk, and not in the RAM. If it has to output the dataframe, it will first need to bring all of them into the RAM, stitch them together and then showcase the final dataframe to youAlso, it shows that the dataframe has been divided into 13 chunks - apparent from the `npartitions=13` that we see from the dataframe. But what do these mean?<jupyter_code>%%time
taxi_data.visualize()<jupyter_output>CPU times: user 7.07 ms, sys: 6.44 ms, total: 13.5 ms
Wall time: 565 ms<jupyter_text>This represents a task graph in Dask which represents parallel computations. - Circles represent functions or tasks- Squares represent results It can be seen that creating a Dask dataframe requires 13 tasks & there are 13 corresponding outputs.Let's do some computation with the dataframe.<jupyter_code>taxi_data.max().visualize()<jupyter_output><empty_output><jupyter_text>It seems like the graph isn't visible - owing to the number of chunks. Let's lower the number of chunks that we will be using for this exercise.<jupyter_code>%%time
taxi_data = dd.read_csv('tripdata.csv', blocksize = '150MB')
taxi_data.visualize()<jupyter_output><empty_output><jupyter_text>Since the file is ~800MB in since, this seems much better. We now have 6 chunks - let's do some computation now<jupyter_code>%%time
taxi_data.max().visualize()<jupyter_output>CPU times: user 12.4 ms, sys: 9.03 ms, total: 21.4 ms
Wall time: 631 ms<jupyter_text>As seen in the graph - instead of computing the max value of each column, calling `.max()` just appended 6 additonal tasks to the original task graph. Two steps were addeed here:- One to compute the max value of each partition - One to compute the agreegated values from each partitionThe lazy dask graph just mimics a lazy engineer. When given a problem, a competent but lazy engineer will just create the solution graph in his/her mind and let it stay there. Only when forced to provide the results (say by a deadline) will the engineer perform the computations. To get the results from the dask graph, we have to force it to perform the computations. We do that by calling the `.compute()` method.<jupyter_code>%%time
taxi_data.max().compute()<jupyter_output>CPU times: user 21.5 s, sys: 3.44 s, total: 24.9 s
Wall time: 18 s<jupyter_text>Let's see how long this took to compute - seems like the entire process took ~13 seconds.Now - let's try doing some _intense_ computation on the data and have a look at the Task Graph<jupyter_code>taxi_data.query('trip_distance <= 50').round({'trip_distance': 0}).groupby(['passenger_count', 'trip_distance']).mean().visualize()<jupyter_output><empty_output><jupyter_text>As you can see, the 6 initial tasks we had for `read_csv` have more tasks representing the aggregate operations that we are trying to do**Hint:** Open the image in a new tab and zoom in to see the graph more clearlyAgain, talking about lazy engineers - there is just a dask graph, we have to force it to perform the computation by calling the `.compute()` method<jupyter_code>%%time
taxi_data.query('trip_distance <= 50').round({'trip_distance': 0}).groupby(['passenger_count', 'trip_distance']).mean().compute()<jupyter_output>CPU times: user 23.8 s, sys: 4.83 s, total: 28.6 s
Wall time: 19.5 s<jupyter_text>This computation took close to 17 seconds to complete. We wouldn't be able to fully appreciate the power of Dask unless we compare this to how much all these computations would normally take with a `pandas` dataframe.Let's read the same file in a pandas dataframe and make a comparision.<jupyter_code>%%time
pd_taxi_data = pd.read_csv('tripdata.csv')<jupyter_output>CPU times: user 14.5 s, sys: 1.64 s, total: 16.1 s
Wall time: 16.7 s<jupyter_text>Reading the dataframe using pandas took a whopping ~15s meanwhile it took only **20.5 ms** with Dask! Let's do all the computations we did earlier using pandas and compare.<jupyter_code>%%time
pd_taxi_data.query('trip_distance <= 50').round({'trip_distance': 0}).groupby(['passenger_count', 'trip_distance']).mean()<jupyter_output>CPU times: user 2.65 s, sys: 1.26 s, total: 3.91 s
Wall time: 4.1 s<jupyter_text>Not quite what we were expecting - looks like Pandas was much faster than Dask. - 4.79s using Pandas- 17s using Dask Maybe our query wasn't complex enough? Let's try with a more complex query?<jupyter_code>def roundup(x, base: int = 5):
    """Round `x` up to nearest `base`"""
    return int(math.ceil(x / float(base))) * base



def transform_dask_dataframe(df: dd.DataFrame) -> dd.DataFrame:
    """Process NYC taxi data"""
    return (
        df[[
            'tpep_pickup_datetime', 'tpep_dropoff_datetime',
            'trip_distance', 'total_amount'
        ]]
        .astype({
            'tpep_pickup_datetime': 'datetime64[ms]',
            'tpep_dropoff_datetime': 'datetime64[ms]'
        })
        .assign(drive_time=(lambda df: (
            df.tpep_dropoff_datetime - df.tpep_pickup_datetime).dt.seconds
            // 300))
        .assign(drive_time=lambda df: df.drive_time)
        .assign(trip_distance=lambda df: df.trip_distance)
        .query('drive_time <= 120 & trip_distance <= 50')
        .drop(['tpep_pickup_datetime', 'tpep_dropoff_datetime'], axis=1)
        .round({'trip_distance': 0})
        .groupby(['drive_time', 'trip_distance'])
        .mean()
        .rename(columns={'total_amount': 'avg_amount'})
    )
def pd_roundup(x, base: int = 5):
    """Round `x` up to nearest `base`"""
    return int(math.ceil(x / float(base))) * base


def transform_pandas_dataframe(df: pd.DataFrame) -> pd.DataFrame:
    """Process NYC taxi data"""
    return (
        df[[
            'tpep_pickup_datetime', 'tpep_dropoff_datetime',
            'trip_distance', 'total_amount'
        ]]
        .astype({
            'tpep_pickup_datetime': 'datetime64[ms]',
            'tpep_dropoff_datetime': 'datetime64[ms]'
        })
        .assign(drive_time=(lambda df: (
            df.tpep_dropoff_datetime - df.tpep_pickup_datetime).dt.seconds
            // 300))
        .assign(drive_time=lambda df: df.drive_time)
        .assign(trip_distance=lambda df: df.trip_distance)
        .query('drive_time <= 120 & trip_distance <= 50')
        .drop(['tpep_pickup_datetime', 'tpep_dropoff_datetime'], axis=1)
        .round({'trip_distance': 0})
        .groupby(['drive_time', 'trip_distance'])
        .mean()
        .rename(columns={'total_amount': 'avg_amount'})
    )<jupyter_output><empty_output><jupyter_text>Let's take it up a notch and work with more data```aws s3 cp "s3://nyc-tlc/trip data/yellow_tripdata_2018-05.csv" "tripdata2.csv"aws s3 cp "s3://nyc-tlc/trip data/yellow_tripdata_2018-06.csv" "tripdata3.csv"aws s3 cp "s3://nyc-tlc/trip data/yellow_tripdata_2018-07.csv" "tripdata4.csv"```<jupyter_code># Read all 4 files
mul_taxi_data = dd.read_csv('tripdata*.csv', blocksize = '500MB')
mul_taxi_data.visualize()
%%time
transform_dask_dataframe(mul_taxi_data).compute()
# Read all 4 files
mul_pd_taxi_data = pd.concat(map(pd.read_csv, glob.glob(os.path.join('', "tripdata*.csv"))))
%%time
transform_pandas_dataframe(mul_pd_taxi_data)<jupyter_output>CPU times: user 28.8 s, sys: 20.8 s, total: 49.6 s
Wall time: 1min 3s
{'fraction_non_alphanumeric': 0.1087151841868823, 'fraction_numerical': 0.025413939160569887, 'mean_word_length': 4.315143246930423, 'pattern_counts': {'":': 0, '<': 52, '<?xml version=': 0, '>': 47, 'https://': 1, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'chain_length': '13', 'dir': 'jupyter-structured-clean-dedup', 'id': '347197', 'license': 'MIT', 'n_tokens_mistral': 2722, 'n_tokens_neox': 2478, 'n_words': 832, 'path': 'dask/dask_lab.ipynb', 'repo_name': 'akashmdubey/Big-Data-Intelligent-Analytics', 'size': '7791'}