"""
# Dask Delayed

These functions do simple operations like add two numbers together, but they sleep for a random amount of time to simulate real work.
"""
import time
import random
import dask


@dask.delayed
def inc(x):
    time.sleep(random.random())
    return x + 1

@dask.delayed
def dec(x):
    time.sleep(random.random())
    return x - 1

def add(x, y):
    time.sleep(random.random())
    return x + y
%%time
x = inc(1)
y = dec(2)
z = add(x, y)
z
%%time
z.compute()
z.visualize()
"""
# Dask natively scales Python

Dask provides advanced parallelism for analytics, enabling performance at scale for the tools you love
"""
import os
import time


import dask
import dask.array as da
import dask.dataframe as dd
import numpy as np
print(dask.__version__)
print(np.__version__)
for dirname, _, filenames in os.walk('/kaggle/input'):
    for filename in filenames:
        print(os.path.join(dirname, filename))
"""
# Reading Dask DataFrame

Read CSV files into a Dask.DataFrame: This parallelizes the `pandas.read_csv()`:
- `blocksize=25e7`: 250MB chunks
- `dtype`: Dask don't explore all the data, sometime it mismatch the dtypes
"""
dtype = {
    'fine_grained_location': 'float64', 
    'officer_id':'object', 
    'county_fips': 'float64'
}

ddf = dd.read_csv(
    "/kaggle/input/stanford-open-policing-project-texas/TX_2010_onwards.csv", 
#     blocksize="25MB", 
    dtype=dtype, 
    low_memory=False,
    assume_missing=True
)

# ddf = ddf.map_partitions(cudf.from_pandas)  # convert pandas partitions into cudf partitions

ddf
%%time
ddf.head()
"""
# What is this Dask Dataframe?
A large, virtual dataframe divided along the index into multiple Pandas dataframes:
"""
ddf.map_partitions(len).compute()
%%time
ddf.map_partitions(type).compute()
"""
# Dask DataFrame Shape
"""
%%time
len(ddf)
%%time
ddf.shape[0].compute()
ddf_rows = ddf.shape[0].compute()
ddf_col = ddf.shape[1]
print(f"Number of rows: {ddf_rows}")
print(f"Number of columns: {ddf_col}")
"""
# Missing Values
"""
%%time
(ddf.isna().sum().compute() / ddf_rows) * 100
"""
# Drop Columns
"""
columns = [
    'police_department', 'driver_age_raw', 'driver_age', 'search_type_raw', 
    'search_type', 'is_arrested'
]
print(f"Number of columns before removing columns: {ddf.shape[1]}")

ddf = ddf.drop(columns, axis=1)

print(f"Number of columns After removing columns: {ddf.shape[1]}")
(ddf.isna().sum().compute() / ddf_rows) * 100
"""
# Drop Rows
"""
print(ddf_rows)
ddf = ddf.dropna(how='any')
print(ddf.shape[0].compute())
(ddf.isna().sum().compute() / ddf_rows) * 100
"""
# Drop_duplicates Rows
"""
# print(ddf.shape[0].compute())
# ddf = ddf.drop_duplicates()
# print(ddf.shape[0].compute())
# print(ddf.shape[0].compute())
# ddf = ddf.drop_duplicates(subset=['id'])
# print(ddf.shape[0].compute())
ddf['id'].nunique().compute()
"""
# Dask DataFrame info
"""
non_object_col = ddf.columns[(ddf.dtypes != object) & (ddf.dtypes != bool)].to_list()
ddf[non_object_col]
ddf.info()
# Describe shows only numerical columns
ddf[non_object_col].describe(percentiles=[.25, .5, .75, .85, .9]).compute()
"""
# Analysing some columns
"""
# for column in ddf.columns:
#     print(f"{column}: Number of unique values {ddf[column].nunique().compute()}")
#     print("_______________________________________________\n")
ddf.columns
ddf['contraband_found'].value_counts().compute()
ddf['stop_outcome'].value_counts().compute()
ddf['driver_race'].value_counts().compute()
ddf['driver_gender'].value_counts().compute()
ddf['state'].value_counts().compute()
# State contains only one value
ddf = ddf.drop('state', axis=1)
ddf['county_name'].value_counts().compute()
montgomery = ddf[ddf['county_name'].str.contains('Montgomery')]
montgomery_county_fips = montgomery.groupby('driver_race').driver_gender.agg(['count'])
montgomery.groupby('driver_race').driver_gender.count().compute()
# montgomery_county_fips.visualize()
# montgomery_county_fips.compute()
"""
# Dask Arrays
"""
import numpy as np
import dask.array as da
a_np = np.arange(1, 50, 3)
a_np
a_da = da.arange(1, 50, 3, chunks=5)
a_da
print(a_da.dtype)
print(a_da.shape)
a_da.visualize()
(a_da ** 2).visualize()
print(a_da.chunks)
print(a_da.chunksize)
x = da.random.random(20, chunks=5)
x
result = x.sum()
result
result.visualize()
result.compute()
x = da.random.random(size=(15, 15), chunks=(10, 5))
x
print(x.chunks)
print(x.chunksize)
result = (x + x.T).sum()
result
result.visualize()
result.compute()
x = da.random.random(size=(20_000, 20_000), chunks=(2_000, 2_000))
x
result = (x + x.T).sum()
result
result.compute()