<jupyter_start><jupyter_text>Create datasets for machine learning- clean data- add unique id to each observation- create time-based features- select features and label for each ML task- split data (last 6 months of 2019 as test data)**ML tasks**- `total_amount`: regression, all observations- `tip_percent`: regression, calculate from `tip_amount`/`total_amount`<jupyter_code>import dask.dataframe as dd
import os
import numpy as np

if 'TAXI_S3' not in os.environ:
    raise ValueError('Set TAXI_S3 environment variable to an S3 location that you have read/write access to')
taxi_path = os.environ['TAXI_S3']
from dask.distributed import Client
from dask_saturn import SaturnCluster
import time

cluster = SaturnCluster(n_workers=10, scheduler_size='xlarge', worker_size='8xlarge', nthreads=32)
client = Client(cluster)

cluster
taxi = dd.read_parquet(f'{taxi_path}/data/taxi_parquet', engine='pyarrow')
taxi.head()
%%time
len(taxi)<jupyter_output>CPU times: user 123 ms, sys: 8.52 ms, total: 132 ms
Wall time: 12.5 s<jupyter_text>Clean data<jupyter_code># only use zones for ML tasks
# TODO: change to spatial join to pull in old data
taxi = taxi[taxi.pickup_taxizone_id.notnull() & taxi.dropoff_taxizone_id.notnull()]

# remove zero fare amount because that causes problems with tip fractions
taxi = taxi[taxi.fare_amount > 0]

# remove tip outliers
taxi['tip_fraction'] = taxi.tip_amount / taxi.fare_amount
tip_outlier = taxi.tip_fraction.mean().compute() + (taxi.tip_fraction.std() * 3)
taxi = taxi[taxi.tip_fraction < tip_outlier]  
np.round(taxi[['tip_amount', 'fare_amount', 'tip_fraction']].describe().compute(), 3).T<jupyter_output><empty_output><jupyter_text>Create some features and split data- These features are stateless, so we can compute them before splitting into train/test sets    - This is also very helpful when deploying the models, as we can take in raw data and return back predictions without passing through a transformation model    - Can you think of more features that might make our models better?- Use last 6 months of 2019 as test set, all other data for train set<jupyter_code>taxi['pickup_weekday'] = taxi.pickup_datetime.dt.weekday
taxi['pickup_weekofyear'] = taxi.pickup_datetime.dt.weekofyear
taxi['pickup_hour'] = taxi.pickup_datetime.dt.hour
taxi['pickup_minute'] = taxi.pickup_datetime.dt.minute
taxi['pickup_week_hour'] = (taxi.pickup_weekday * 24) + taxi.pickup_hour
features = [
    'pickup_datetime',  # timestamps won't be used for training models
    'dropoff_datetime', # included for downstream analysis
    'pickup_taxizone_id',
    'dropoff_taxizone_id',
    'pickup_weekday',
    'pickup_weekofyear',
    'pickup_hour',
    'pickup_minute',
    'pickup_week_hour',
    'passenger_count',
]
import datetime

# this currently pulls from 2017, because zone columns are empty before then
# TODO: add spatial join to convert old lat/lon data to zones
train = taxi[(taxi.pickup_datetime >= datetime.datetime(2009, 1, 1)) &
             (taxi.pickup_datetime < datetime.datetime(2019, 7, 1))]
test = taxi[(taxi.pickup_datetime >= datetime.datetime(2019, 7, 1)) & 
            (taxi.pickup_datetime < datetime.datetime(2020, 1, 1))]
from dask import compute

compute(
    train.shape[0],
    train.pickup_datetime.min(),
    train.pickup_datetime.max(),
    
    test.shape[0],
    test.pickup_datetime.min(),
    test.pickup_datetime.max(),
)<jupyter_output><empty_output><jupyter_text>Write out files**NOTE**: each ML dataset will generate different uuid's, so will not be able to link between them- `amount`: predict total amount- `tip`: predict tip percentage<jupyter_code>import s3fs
import uuid

fs = s3fs.S3FileSystem()
ml_path = f'{taxi_path}/data/ml'


def write_df(df: dd.DataFrame, path: str, partition_size='100MB', rm=True, create_id=True) -> dd.DataFrame:
    """
    Add uuid, repartition, and write dataframe to parquet
        
    Parameters:
        df (DataFrame): DataFrame to write
        path (str): S3 Path to write to
        partition_size (str): size for repartitioning
        rm (bool): Whether to remove files before writing
        create_id (bool): Whether to add a unique id column
    """
    if rm and fs.exists(path):
        fs.rm(path, recursive=True)
    
    if create_id:
        cols = df.columns.values.tolist()
        df['id'] = df.iloc[:, 0].apply(lambda _: uuid.uuid4().hex, meta=('id', 'object'))
        df = df[['id'] + cols]
    
    if partition_size:
        df = df.repartition(partition_size=partition_size)
    
    df.to_parquet(path, write_index=False, engine='pyarrow', compression='snappy')<jupyter_output><empty_output><jupyter_text>Fare amount prediction, use all data<jupyter_code>amt_train = train[features + ['fare_amount']]
amt_test = test[features + ['fare_amount']]
%time write_df(amt_train, f'{ml_path}/amount_train')
%time write_df(amt_test, f'{ml_path}/amount_test')<jupyter_output>CPU times: user 5.55 s, sys: 27.1 ms, total: 5.58 s
Wall time: 7min 47s<jupyter_text>Tip percentage prediction, only use credit card payment`payment_type` of 1 means credit card, based on [data dictionary](https://www1.nyc.gov/assets/tlc/downloads/pdf/data_dictionary_trip_records_yellow.pdf)<jupyter_code>tip_train = train[(train.payment_type == 1) & train.tip_fraction.notnull()][features + ['tip_fraction']]
tip_test = test[(test.payment_type == 1) & test.tip_fraction.notnull()][features + ['tip_fraction']]
%time write_df(tip_train, f'{ml_path}/tip_train')
%time write_df(tip_test, f'{ml_path}/tip_test')<jupyter_output>CPU times: user 5.32 s, sys: 32.3 ms, total: 5.35 s
Wall time: 3min 26s<jupyter_text>Write samples for single-node work<jupyter_code>%%time

tip_train_sample = (
    dd.read_parquet(f'{ml_path}/tip_train', engine='pyarrow')
    .sample(frac=0.05, replace=False, random_state=42)
)
write_df(tip_train_sample, f'{ml_path}/tip_train_sample', create_id=False)

tip_test_sample = (
    dd.read_parquet(f'{ml_path}/tip_test', engine='pyarrow')
    .sample(frac=0.05, replace=False, random_state=42)
)
write_df(tip_test_sample, f'{ml_path}/tip_test_sample', create_id=False)<jupyter_output>CPU times: user 1.61 s, sys: 43.7 ms, total: 1.65 s
Wall time: 49.2 s<jupyter_text>Check our work(might need to restart kernel, `pyarrow` gets a bit confused)<jupyter_code>amt_train_ = dd.read_parquet(f'{ml_path}/amount_train', engine='pyarrow')
len(amt_train_)
amt_train_.head()
%time np.round(amt_train_.describe().compute(), 3).T
amt_test_ = dd.read_parquet(f'{ml_path}/amount_test', engine='pyarrow')
len(amt_test_)
amt_test_.head()
%time np.round(amt_test_.describe().compute(), 3).T
tip_train_ = dd.read_parquet(f'{ml_path}/tip_train', engine='pyarrow')
len(tip_train_)
tip_train_.head()
%time np.round(tip_train_.describe().compute(), 3).T
tip_test_ = dd.read_parquet(f'{ml_path}/tip_test', engine='pyarrow')
len(tip_test_)
tip_test_.head()
%time np.round(tip_test_.describe().compute(), 3).T<jupyter_output>CPU times: user 959 ms, sys: 7.65 ms, total: 967 ms
Wall time: 5.41 s
{'fraction_non_alphanumeric': 0.1246065808297568, 'fraction_numerical': 0.02017167381974249, 'mean_word_length': 5.264336917562724, 'pattern_counts': {'":': 0, '<': 31, '<?xml version=': 0, '>': 32, 'https://': 1, 'lorem ipsum': 0, 'www.': 1, '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': '244143', 'license': 'BSD-3-Clause', 'n_tokens_mistral': 2532, 'n_tokens_neox': 2395, 'n_words': 603, 'path': 'taxi_demo/etl/ml_datasets.ipynb', 'repo_name': 'mmccarty/saturn-cloud-examples', 'size': '6990'}