<jupyter_start><jupyter_text>NYC_Taxi Fare Prediction with RayDP<jupyter_code>import ray
import os
import datetime
import pandas as pd, numpy as np

import torch
import torch.nn as nn
import torch.nn.functional as F

from pyspark.sql.functions import *

from raydp.spark import context
from raydp.spark.torch.estimator import TorchEstimator
from raydp.spark.utils import random_split<jupyter_output><empty_output><jupyter_text>Initialize or connect to existed Ray cluster<jupyter_code># Firstly, You need to init or connect to a ray cluster. Note that you should set include_java to True.
# For more config info in ray, please refer the ray doc. https://docs.ray.io/en/latest/package-ref.html
# ray.init(address="auto", redis_password="<PASSWORD>")
ray.init(include_java=True)<jupyter_output><empty_output><jupyter_text>Create Spark session based on RayDP API<jupyter_code># After initialize ray cluster, you can use the raydp api to get a spark session
app_name = "NYC_Taxi Fare Prediction with RayDP"
num_executors = 2
cores_per_executor = 1
memory_per_executor = "5GB"
spark = context.init_spark(app_name, num_executors, cores_per_executor, memory_per_executor)<jupyter_output><empty_output><jupyter_text>Distributed data preprocessing with pyspark<jupyter_code># Then you can operate as you are using spark
# The dataset can be downloaded from https://www.kaggle.com/c/new-york-city-taxi-fare-prediction/data
data = spark.read.format("csv").option("header", "true") \
        .option("inferSchema", "true") \
        .load("/mnt/DP_disk8/nyc_taxi.csv")
# filter data
data = data.select("pickup_longitude", "pickup_latitude", "dropoff_longitude", "dropoff_latitude", "passenger_count", "fare_amount") \
    .filter(data['pickup_longitude'] <= -73.0)  \
    .filter(data['pickup_longitude'] >= -74.3)  \
    .filter(data['dropoff_longitude'] <= -73.0) \
    .filter(data['dropoff_longitude'] >= -74.3) \
    .filter(data['pickup_latitude'] <= 41.7)    \
    .filter(data['pickup_latitude'] >= 40.6)    \
    .filter(data['dropoff_latitude'] <= 41.7)   \
    .filter(data['dropoff_latitude'] >= 40.6)   \
    .filter(data['passenger_count'] <= 10)      \
    .filter(data['fare_amount'] >= 0.0)         \
    .filter(abs(data['dropoff_longitude'] - data['pickup_longitude']) <= 5) \
    .filter(abs(data['dropoff_latitude'] - data['pickup_latitude']) <= 5)
# add two new features
data = data.withColumn("abs_diff_longitude", abs(col("dropoff_longitude") - col("pickup_longitude"))) \
           .withColumn("abs_diff_latitude", abs(col("dropoff_latitude") - col("pickup_latitude")))
# split data into train_dataset and test_dataset
train_df, test_df = random_split(data, [0.7, 0.3])
# train_df = train_data_select.sample(False, 0.0005)
# test_df = train_data_select.sample(False, 0.0001)
train_df.show()
test_df.show()<jupyter_output><empty_output><jupyter_text>Distributed model training and evaluation<jupyter_code># Define the model, loss function and optimizer
class NYC_Model(nn.Module):
    def __init__(self):
        super(NYC_Model, self).__init__()
        
        self.fc1 = nn.Linear(7, 14)
        self.fc2 = nn.Linear(14, 7)
        self.fc3 = nn.Linear(7, 1)

    def forward(self, x):
        
        x = F.relu(self.fc1(x))
        x = F.relu(self.fc2(x))
        x = self.fc3(x)
        return x

nyc_model = NYC_Model()
criterion = nn.MSELoss()
optimizer = torch.optim.SGD(nyc_model.parameters(), lr=0.001)
# create the estimator which is a scikit-learn like API for Torch distributed model training
features = [field.name for field in list(train_data_select.schema) if field.name != "fare_amount"]
estimator = TorchEstimator(num_workers=4,
                           model=nyc_model,
                           optimizer=optimizer,
                           loss=criterion, 
                           feature_columns=features,
                           label_column="fare_amount",
                           batch_size=100,
                           num_epochs=100)
# distributed training the model
estimator.fit(train_df)
# evaluate the model
estimator.evaluate(test_df)
# shutdown raydp and ray
estimator.shutdown()
context.stop_spark()
ray.shutdown()<jupyter_output><empty_output>
{'fraction_non_alphanumeric': 0.12280285035629454, 'fraction_numerical': 0.017339667458432306, 'mean_word_length': 3.763574660633484, 'pattern_counts': {'":': 0, '<': 29, '<?xml version=': 0, '>': 27, 'https://': 2, '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': '5', 'dir': 'jupyter-structured-clean-dedup', 'id': '472252', 'license': 'Apache-2.0', 'n_tokens_mistral': 1472, 'n_tokens_neox': 1411, 'n_words': 303, 'path': 'examples/NYC_Taxi.ipynb', 'repo_name': 'buaazhwb/oap-raydp', 'size': '4203'}