<jupyter_start><jupyter_text>Scaling Machine Learning in Python Single-node workflowWe'll first start off with a typical data preparation and machine learning workflow utilizing only the Jupyter Server. Monitor resource utilizationFor this workshop it's important to monitor CPU and memory utilization when running various commands. It will help with understanding which operations are slow - and which ones run faster on a cluster!To monitor resource utilization of the Jupyter Server, open a new Terminal window inside Jupyter Lab and run `htop`. You can position the window to view the notebook and terminal on the same screen:![htop](img/htop.png)  Load dataThis workshop will utilize data for yellow taxi rides from the 2019 calendar year. The machine learning exercises involve predicting the "tip fraction" of each ride - how much a rider will tip the driver as a fraction of the charged fare amount.Let's operate with one month for now to explore the data and build out the machine learning code.> Note: `s3.open()` is only required because we're using an anonymous S3 connection. If you have AWS credentials set up, you could pass in just the S3 URL string to `pd.read_csv()`.<jupyter_code>import s3fs
import pandas as pd
import numpy as np
import warnings
warnings.simplefilter("ignore")

s3 = s3fs.S3FileSystem(anon=True)
jan2019_file = 's3://nyc-tlc/trip data/yellow_tripdata_2019-01.csv'
jan2019_size = s3.du(jan2019_file)
print(f"Size: {round(jan2019_size / 1e6, 2)} MB")
%%time

taxi = pd.read_csv(
    s3.open(jan2019_file, mode='rb',),
    parse_dates=['tpep_pickup_datetime', 'tpep_dropoff_datetime']
)
taxi.head()<jupyter_output><empty_output><jupyter_text>ExerciseHow many rows are in the `taxi` DataFrame?<jupyter_code><FILL IN>
len(taxi)<jupyter_output><empty_output><jupyter_text>Memory usage is also an important consideration, as DataFrames often take more space in memory than on disk:<jupyter_code>taxi_bytes = taxi.memory_usage(deep=True).sum()
print(f"Size (MB): {taxi_bytes / 1e6}")<jupyter_output><empty_output><jupyter_text>Exploratory analysisFor this workshop, we will just look at column statistics. There are many more explorary analyses that can performed with `pandas` and data visualization tools<jupyter_code>%%time
np.round(taxi.describe().T, 3)<jupyter_output><empty_output><jupyter_text>Feature engineeringWe are using stateless features, meaning the features values for a given observation don't depend on other observations. This is allows us to create features before performing any data splitting.Then, split data into train/test sets.<jupyter_code># specify feature and label column names
raw_features = [
    'tpep_pickup_datetime', 
    'passenger_count', 
    'tip_amount', 
    'fare_amount',
]
features = [
    'pickup_weekday', 
    'pickup_weekofyear', 
    'pickup_hour', 
    'pickup_week_hour', 
    'pickup_minute', 
    'passenger_count',
]
label = 'tip_fraction'
def prep_df(taxi_df):
    '''
    Generate features from a raw taxi dataframe.
    '''
    df = taxi_df[taxi_df.fare_amount > 0][raw_features].copy()  # avoid divide-by-zero
    df[label] = df.tip_amount / df.fare_amount
     
    df['pickup_weekday'] = df.tpep_pickup_datetime.dt.weekday
    df['pickup_weekofyear'] = df.tpep_pickup_datetime.dt.weekofyear
    df['pickup_hour'] = df.tpep_pickup_datetime.dt.hour
    df['pickup_week_hour'] = (df.pickup_weekday * 24) + df.pickup_hour
    df['pickup_minute'] = df.tpep_pickup_datetime.dt.minute
    df = df[features + [label]].astype(float).fillna(-1)
    
    return df
taxi_feat = prep_df(taxi)
taxi_feat.head()<jupyter_output><empty_output><jupyter_text>Split into train/test sets<jupyter_code>from sklearn.model_selection import train_test_split

X_train, X_test, y_train, y_test = train_test_split(
    taxi_feat[features], 
    taxi_feat[label], 
    test_size=0.3,
    random_state=42
)
X_train.shape, y_train.shape
X_test.shape, y_test.shape<jupyter_output><empty_output><jupyter_text>Train modelWe'll train a linear model to predict `tip_fraction`. We define a `Pipeline` to encompass both feature scaling and model training. This will be useful later when performing a grid search.Evaluate the model against the test set using RMSE. We'll also save out the model for later use.<jupyter_code>from sklearn.pipeline import Pipeline
from sklearn.linear_model import ElasticNet

from sklearn.preprocessing import StandardScaler
from sklearn.metrics import mean_squared_error

pipeline = Pipeline(steps=[
    ('scale', StandardScaler()),
    ('clf', ElasticNet(normalize=False, max_iter=100, l1_ratio=0)),
])
%%time
fitted = pipeline.fit(X_train, y_train)
%%time
preds = fitted.predict(X_test)
mean_squared_error(y_test, preds, squared=False)
import cloudpickle

with open('/tmp/model.pkl', 'wb') as f:
    cloudpickle.dump(fitted, f)<jupyter_output><empty_output><jupyter_text>Hooray!We trained a terrible model. But that's okay. The point of this workshop is to scale our work, not make the model perfect! Let's step things upWe were able to train a single, simple model on a sample of the taxi data (single month from 2019). In most model building settings more data and more compute would be required. We will explore the following scenarios where the single-node environment has challenges - and see how Dask solves these problems!- Tune hyperparameters (train numerous models)- Load and process large dataset (full year's data)- Predict over large dataset- Train model with large dataset  Tune hyperparametersOne model can be trained comfortably and fairly quickly with the amount of memory and processing power the Jupyter Server has. A hyperparameter search involves training many models - which means it needs a lot of processing power to finish in a reasonable amount of time.Scikit-learn's `GridSearchCV` accepts an `n_jobs` argument, which will train `n` different models at the same time. Setting it to `-1` uses all the CPU cores in this Jupyter Server. Watch `htop` here - the cores will be saturated while memory will not be.<jupyter_code>from sklearn.model_selection import GridSearchCV

params = {
    'clf__l1_ratio': np.arange(0, 1.1, 0.1),
    'clf__alpha': [0, 0.5, 1, 2],
}

grid_search = GridSearchCV(
    pipeline, 
    params, 
    cv=3, 
    n_jobs=-1, 
    verbose=1, 
    scoring='neg_mean_squared_error',
)<jupyter_output><empty_output><jupyter_text>Take a sample so this portion will execute faster<jupyter_code>taxi_sample = taxi_feat.sample(frac=0.1, replace=False)
%%time
_ = grid_search.fit(
    taxi_sample[features],
    taxi_sample[label],
)<jupyter_output><empty_output><jupyter_text>This works, but...This is a pretty small grid search. With a larger grid, it would take quite a while.![waiting](https://media3.giphy.com/media/QBd2kLB5qDmysEXre9/giphy.gif?cid=ecf05e47t2wdryi59apjgwa3okzhxpctvxjsqaxautawn96i&rid=giphy.gif)---   Load and process large datasetLet's look at the size of the files on disk using `s3fs`.<jupyter_code>files = s3.glob('s3://nyc-tlc/trip data/yellow_tripdata_2019-*.csv')
total_size = 0
for f in files:
    size = s3.du(f)
    total_size += size
    
    print(f"{f}, Size: {round(size / 1e6, 2)} MB")
print()
print(f"Total size: {round(total_size / 1e9, 2)} GB")<jupyter_output><empty_output><jupyter_text>Files end up taking more space when loaded into memory, but let's see if this will fit on our Jupyter Server. We can loop through the files and concatenate them into one DataFrame. Watch memory utilization as the loop runs!<jupyter_code>def load_csv(file):
    df = pd.read_csv(
        s3.open(file, mode='rb'),
        parse_dates=['tpep_pickup_datetime', 'tpep_dropoff_datetime']
    )
    return df
%%time

dfs = []
for f in files:
    print(f)
    df = load_csv(f)
    print(f'{len(df)} rows, {df.memory_usage(deep=True).sum() / 1e6} MB')
    dfs.append(df)
taxi_big = pd.concat(dfs)<jupyter_output><empty_output><jupyter_text>![explosion](https://media4.giphy.com/media/13d2jHlSlxklVe/giphy.gif)Oh no! Looks like we have enough memory to load the CSV files individually, but not to concatenate them together into one DataFrame.---   Predict over large datasetWe can't load a large dataset into one DataFrame, but if we needed to predict over a large dataset we could batch it and collect the results.> Note: If your kernel died when trying to concatenate the big DataFrame, you will need to re-run all the preceding cells in this notebook - just not the one that killed the kernel!This will work, but watch `htop` - only one core will be used at a time.<jupyter_code>%%time

predicted_dfs = []
for f in files:
    print(f)
    df = load_csv(f)
    feat_df = prep_df(df)
    predicted = fitted.predict(feat_df[features])
    predicted_df = pd.DataFrame(predicted, columns=['prediction'])
    predicted_dfs.append(predicted_df)
    
predicted_full = pd.concat(predicted_dfs)
predicted_full.describe()<jupyter_output><empty_output>
{'fraction_non_alphanumeric': 0.09093978615644345, 'fraction_numerical': 0.013055711873944852, 'mean_word_length': 5.004054054054054, 'pattern_counts': {'":': 0, '<': 50, '<?xml version=': 0, '>': 53, 'https://': 2, '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': '12', 'dir': 'jupyter-structured-clean-dedup', 'id': '341102', 'license': 'BSD-3-Clause', 'n_tokens_mistral': 2948, 'n_tokens_neox': 2811, 'n_words': 935, 'path': '02-single-node.ipynb', 'repo_name': 'tvquynh/workshop-scaling-ml', 'size': '8885'}