<jupyter_start><jupyter_text>Python scientific StackRef: <NAME> - PyCon DE 2017 Keynote  Numeric Python for scientific computing- Actively supported packages for (almost) everything- Trusted and accepted by scientific communityHowever, - Everything is object in Python (fast implementation, slow processing)- Limited to single thread (due to “Global Interpreter Lock (GIL)”)- Limited to in-memory data  Multiprocessing- Multiprocessing is a way to get around limitation in threading in python- If data can be split perfectly into independent chunks (i.e., embarressingly parallel), then each chunk can be processed independently. Then the results are aggregated in reduce phase.Efficiently Exploiting Multiple Cores with Python<jupyter_code># Sequantional processing of an array of data
import math
def func(x):
    return x ** 2 + math.sin(x) + x ** 5

d = range(1, 5_000_000)

_ = list(map(func, d))

# Takes 4.45 seconds to run
# Multiprocessing of an array of data
import multiprocessing

with multiprocessing.Pool(4) as pool:
    __ = pool.map(func, d)

pool.close()
# Takes 1.85 seconds to run<jupyter_output><empty_output><jupyter_text>Big data collectionsApplying SQL like queries on large dataset (both structural and non structural)- Spark- MapReduce- SQL  Task schedulersIt creates a DAG graph between tasks, creates a pipeline on how tasks are executed and which ttaks must be completed before starting a new task - Apache Airflow- Dask   Why choosing dask- Numpy and Pandas are two main tools for data scinetist to explore and clean data- Both can be run on only a single core- For most cases (such as university resercah), size of data is too small so there is not need to have scalable solution; but if facing a big data, scalable approcahes must be explored.  Scalable Pandas DataFramesList of Dask dataframe API- groupby- sample/resample- mean/min/max/std/...- drop/dronna/drop_duplicate- iterrows/itertuples- join/append (but no concat)  Analyzing flight data with dask.dataframedata can be downloaded from here<jupyter_code>cols = ['year', 'month', 'day_of_month', 'day_of_week', 'deptime', 'crs_deptime', 'arrtime', 
        'crs_arrtime', 'unique_carrier', 'flight_num', 'tail_num', 'actual_elapsed_time',
        'crs_elapsed_time', 'air_time', 'arrdelay', 'depdelay', 'origin', 'dest', 'distance', 
        'taxi_in', 'taxi_out', 'cancelled', 'cancellation_code', 'diverted', 'carrier_delay',
        'weather_delay', 'nas_delay', 'security_delay', 'late_aircraft_delay']

dtypes = {'cancellation_code': object, 'taxi_in': float, 'taxi_out': float, 'cancelled': bool,
          'diverted': bool, 'carrier_delay': float, 'weather_delay': float, 'nas_delay': float,
          'security_delay': float, 'late_aircraft_delay': float, 'tail_num': object,
          'crs_deptime': float, 'crs_arrtime': float, 'flight_num': float, 'crs_elapsed_time': float,
          'distance': float}
import dask.dataframe as pd_dask
import dask

df_dask = pd_dask.read_csv('data/*.csv', header=0, names=cols, dtype=dtypes)
df_dask.head()
# in dask
_ = (df_dask
      .groupby(['day_of_week', 'origin'])
      .depdelay
      .aggregate(['mean', 'std']))
_
from dask.diagnostics import visualize
_.visualize()<jupyter_output><empty_output><jupyter_text>The above shows the graph before optimisation before the computation is actually run (by calling the .compute() method). The graph is simplified with several optimization passes to improve efficiency. To visualize the final graph to be run, we can set the optimize_graph keyword as True<jupyter_code>_.visualize(optimize_graph=True)
dask.compute(_)<jupyter_output><empty_output><jupyter_text>Scalable Numpy arrayList of Dask array API<jupyter_code>import dask.array as da

arr = df_dask.depdelay.values

y = da.nanmax(arr.T * arr)
y
y = dask.compute(y)
y<jupyter_output><empty_output><jupyter_text>Scalable Sklearn and JoblibThere are three ways to parallel ML algorithms: use dask as `parallel_backend` in `joblib`<jupyter_code>import joblib
from dask.distributed import Client

client = Client()

from sklearn.datasets import load_digits
from sklearn.linear_model import LogisticRegression

digits = load_digits()
model = LogisticRegression()

with joblib.parallel_backend('dask'):
    model.fit(digits.data, digits.target)<jupyter_output><empty_output><jupyter_text>use dask ML implementation<jupyter_code>from dask_ml.linear_model import LogisticRegression
model = LogisticRegression()

model.fit(digits.data, digits.target)<jupyter_output><empty_output><jupyter_text>use dask parallel meta-estimators ParallelPostFit will parallelize the predict, predict_proba and transform methods, enabling them to work on large (possibly larger-than memory) datasets.<jupyter_code>from sklearn.linear_model import LogisticRegression
from dask_ml.wrappers import ParallelPostFit

model = ParallelPostFit(estimator=LogisticRegression())

model.fit(digits.data, digits.target) # dask doesn nothing here
y_pred = model.predict(digits.data)  # dask is used to predict large dataset (which ca not ne fitted into memory)<jupyter_output><empty_output><jupyter_text>Parallelize Existing Codebases<jupyter_code># Normal implementation
results = []
for x in range(100): 
    for y in range(200): 
        if x < y:
            results.append(x ** 2 + y ** 2)
        else:
            results.append(x ** 3 + y ** 3)
# Implementation in dask
import dask
from dask import delayed

f = delayed(lambda x, y: x ** 2 + y ** 2)
g = delayed(lambda x, y: x ** 3 + y ** 3)

results = []
for x in range(100): 
    for y in range(200): 
        if x < y:
            results.append(f(x,y))
        else:
            results.append(g(x,y))
            
result = dask.compute(results)                        
client<jupyter_output><empty_output>
{'fraction_non_alphanumeric': 0.09622052704576976, 'fraction_numerical': 0.007281553398058253, 'mean_word_length': 4.763236763236764, 'pattern_counts': {'":': 0, '<': 36, '<?xml version=': 0, '>': 34, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 1, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'chain_length': '8', 'dir': 'jupyter-structured-clean-dedup', 'id': '138544', 'license': 'Apache-2.0', 'n_tokens_mistral': 1854, 'n_tokens_neox': 1779, 'n_words': 592, 'path': 'dask-slides.ipynb', 'repo_name': 'nikzadb/dask-tutorial', 'size': '5778'}