dataproc/convert_taxi_to_parquet.py

####################################################################################
# Copyright 2022 Google LLC
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
# 
#     https://www.apache.org/licenses/LICENSE-2.0
# 
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
####################################################################################

# Author:  Adam Paternostro
# Summary: Opens the taxi parquet files and renames the fields as well as partitions the data.

# Note: NYC changes from CSV to Parquet format, this file has been updated on 05/16/2022

from pyspark.sql.dataframe import DataFrame
from pyspark.sql import SparkSession
from pyspark.sql.functions import col, year, month
from pyspark.sql.types import StructType, StructField, StringType, IntegerType, DoubleType, TimestampType
from datetime import datetime
import time
import sys


# All the data cannot be read at once since MergeSchema does not work for the NYC data
def ProcessGreenFile(spark, filenameAndPath):
    df_source = spark.read.option("mergeSchema", "true").parquet(filenameAndPath)

    print("ProcessGreenFile: filenameAndPath: ",filenameAndPath)

    df_TypeCast = df_source \
        .withColumn("Vendor_ID",col("VendorID").cast(IntegerType())) \
        .withColumn("Pickup_DateTime",col("lpep_pickup_datetime").cast(TimestampType())) \
        .withColumn("Dropoff_DateTime",col("lpep_dropoff_datetime").cast(TimestampType())) \
        .withColumn("Store_And_Forward",col("store_and_fwd_flag").cast(StringType())) \
        .withColumn("Rate_Code_Id",col("RatecodeID").cast(IntegerType())) \
        .withColumn("PULocationID",col("PULocationID").cast(IntegerType())) \
        .withColumn("DOLocationID",col("DOLocationID").cast(IntegerType())) \
        .withColumn("Passenger_Count",col("passenger_count").cast(IntegerType())) \
        .withColumn("Trip_Distance",col("trip_distance").cast(DoubleType())) \
        .withColumn("Fare_Amount",col("fare_amount").cast(DoubleType())) \
        .withColumn("Surcharge",col("extra").cast(DoubleType())) \
        .withColumn("MTA_Tax",col("mta_tax").cast(DoubleType())) \
        .withColumn("Tip_Amount",col("tip_amount").cast(DoubleType())) \
        .withColumn("Tolls_Amount",col("tolls_amount").cast(DoubleType())) \
        .withColumn("Ehail_Fee",col("ehail_fee").cast(DoubleType())) \
        .withColumn("Improvement_Surcharge",col("improvement_surcharge").cast(DoubleType())) \
        .withColumn("Total_Amount",col("total_amount").cast(DoubleType())) \
        .withColumn("Payment_Type_Id",col("payment_type").cast(IntegerType())) \
        .withColumn("Trip_Type",col("trip_type").cast(IntegerType())) \
        .withColumn("Congestion_Surcharge",col("congestion_surcharge").cast(DoubleType()))

    df_result = df_TypeCast.select( \
        'Vendor_Id', \
        'Pickup_DateTime', \
        'Dropoff_DateTime', \
        'Store_And_Forward', \
        'Rate_Code_Id', \
        'PULocationID', \
        'DOLocationID', \
        'Passenger_Count', \
        'Trip_Distance', \
        'Fare_Amount', \
        'Surcharge', \
        'MTA_Tax', \
        'Tip_Amount', \
        'Tolls_Amount', \
        'Ehail_Fee', \
        'Improvement_Surcharge', \
        'Total_Amount', \
        'Payment_Type_Id', \
        'Trip_Type', \
        'Congestion_Surcharge' \
    )
   
    return df_result


# All the data cannot be read at once since MergeSchema does not work for the NYC data
def ProcessYellowFile(spark, filenameAndPath):
    df_source = spark.read.parquet(filenameAndPath)

    print("ProcessYellowFile: filenameAndPath: ",filenameAndPath)

    df_TypeCast = df_source \
        .withColumn("Vendor_ID",col("VendorID").cast(IntegerType())) \
        .withColumn("Pickup_DateTime",col("tpep_pickup_datetime").cast(TimestampType())) \
        .withColumn("Dropoff_DateTime",col("tpep_dropoff_datetime").cast(TimestampType())) \
        .withColumn("Passenger_Count",col("passenger_count").cast(IntegerType())) \
        .withColumn("Trip_Distance",col("trip_distance").cast(DoubleType())) \
        .withColumn("Rate_Code_Id",col("RatecodeID").cast(IntegerType())) \
        .withColumn("Store_And_Forward",col("store_and_fwd_flag").cast(StringType())) \
        .withColumn("PULocationID",col("PULocationID").cast(IntegerType())) \
        .withColumn("DOLocationID",col("DOLocationID").cast(IntegerType())) \
        .withColumn("Payment_Type_Id",col("payment_type").cast(IntegerType())) \
        .withColumn("Fare_Amount",col("fare_amount").cast(DoubleType())) \
        .withColumn("Surcharge",col("extra").cast(DoubleType())) \
        .withColumn("MTA_Tax",col("mta_tax").cast(DoubleType())) \
        .withColumn("Tip_Amount",col("tip_amount").cast(DoubleType())) \
        .withColumn("Tolls_Amount",col("tolls_amount").cast(DoubleType())) \
        .withColumn("Improvement_Surcharge",col("improvement_surcharge").cast(DoubleType())) \
        .withColumn("Total_Amount",col("total_amount").cast(DoubleType())) \
        .withColumn("Congestion_Surcharge",col("congestion_surcharge").cast(DoubleType()))
        #.withColumn("Airport_Fee",col("airport_fee").cast(DoubleType()))

    df_result = df_TypeCast.select( \
        'Vendor_Id', \
        'Pickup_DateTime', \
        'Dropoff_DateTime', \
        'Passenger_Count', \
        'Trip_Distance', \
        'Rate_Code_Id', \
        'Store_And_Forward', \
        'PULocationID', \
        'DOLocationID', \
        'Payment_Type_Id', \
        'Fare_Amount', \
        'Surcharge', \
        'MTA_Tax', \
        'Tip_Amount', \
        'Tolls_Amount', \
        'Improvement_Surcharge', \
        'Total_Amount', \
        'Congestion_Surcharge' \
        )
    
    return df_result


def ConvertTaxiData(sourceYellow, sourceGreen, destination):
    print("ConvertTaxiData: sourceYellow: ",sourceYellow)
    print("ConvertTaxiData: sourceGreen:  ",sourceGreen)
    print("ConvertTaxiData: destination:  ",destination)

    spark = SparkSession \
        .builder \
        .appName("ConvertTaxiData") \
        .getOrCreate()
    

    ################################################################################################
    # Yellow
    ################################################################################################    
    yellow_path = sourceYellow.replace("/*/*.parquet","")

    # The NYC taxi data has fields that change data types and some are incompatible with mergeScheme   
    # 2019
    yellow_tripdata_2019_01 = ProcessYellowFile(spark,f"{yellow_path}/2019/yellow_tripdata_2019-01.parquet")
    yellow_tripdata_2019_02 = ProcessYellowFile(spark,f"{yellow_path}/2019/yellow_tripdata_2019-02.parquet")
    yellow_tripdata_2019_03 = ProcessYellowFile(spark,f"{yellow_path}/2019/yellow_tripdata_2019-03.parquet")
    yellow_tripdata_2019_04 = ProcessYellowFile(spark,f"{yellow_path}/2019/yellow_tripdata_2019-04.parquet")
    yellow_tripdata_2019_05 = ProcessYellowFile(spark,f"{yellow_path}/2019/yellow_tripdata_2019-05.parquet")
    yellow_tripdata_2019_06 = ProcessYellowFile(spark,f"{yellow_path}/2019/yellow_tripdata_2019-06.parquet")
    yellow_tripdata_2019_07 = ProcessYellowFile(spark,f"{yellow_path}/2019/yellow_tripdata_2019-07.parquet")
    yellow_tripdata_2019_08 = ProcessYellowFile(spark,f"{yellow_path}/2019/yellow_tripdata_2019-08.parquet")
    yellow_tripdata_2019_09 = ProcessYellowFile(spark,f"{yellow_path}/2019/yellow_tripdata_2019-09.parquet")
    yellow_tripdata_2019_10 = ProcessYellowFile(spark,f"{yellow_path}/2019/yellow_tripdata_2019-10.parquet")
    yellow_tripdata_2019_11 = ProcessYellowFile(spark,f"{yellow_path}/2019/yellow_tripdata_2019-11.parquet")
    yellow_tripdata_2019_12 = ProcessYellowFile(spark,f"{yellow_path}/2019/yellow_tripdata_2019-12.parquet")

    # 2020
    yellow_tripdata_2020_01 = ProcessYellowFile(spark,f"{yellow_path}/2020/yellow_tripdata_2020-01.parquet")
    yellow_tripdata_2020_02 = ProcessYellowFile(spark,f"{yellow_path}/2020/yellow_tripdata_2020-02.parquet")
    yellow_tripdata_2020_03 = ProcessYellowFile(spark,f"{yellow_path}/2020/yellow_tripdata_2020-03.parquet")
    yellow_tripdata_2020_04 = ProcessYellowFile(spark,f"{yellow_path}/2020/yellow_tripdata_2020-04.parquet")
    yellow_tripdata_2020_05 = ProcessYellowFile(spark,f"{yellow_path}/2020/yellow_tripdata_2020-05.parquet")
    yellow_tripdata_2020_06 = ProcessYellowFile(spark,f"{yellow_path}/2020/yellow_tripdata_2020-06.parquet")
    yellow_tripdata_2020_07 = ProcessYellowFile(spark,f"{yellow_path}/2020/yellow_tripdata_2020-07.parquet")
    yellow_tripdata_2020_08 = ProcessYellowFile(spark,f"{yellow_path}/2020/yellow_tripdata_2020-08.parquet")
    yellow_tripdata_2020_09 = ProcessYellowFile(spark,f"{yellow_path}/2020/yellow_tripdata_2020-09.parquet")
    yellow_tripdata_2020_10 = ProcessYellowFile(spark,f"{yellow_path}/2020/yellow_tripdata_2020-10.parquet")
    yellow_tripdata_2020_11 = ProcessYellowFile(spark,f"{yellow_path}/2020/yellow_tripdata_2020-11.parquet")
    yellow_tripdata_2020_12 = ProcessYellowFile(spark,f"{yellow_path}/2020/yellow_tripdata_2020-12.parquet")

    # 2021
    yellow_tripdata_2021_01 = ProcessYellowFile(spark,f"{yellow_path}/2021/yellow_tripdata_2021-01.parquet")
    yellow_tripdata_2021_02 = ProcessYellowFile(spark,f"{yellow_path}/2021/yellow_tripdata_2021-02.parquet")
    yellow_tripdata_2021_03 = ProcessYellowFile(spark,f"{yellow_path}/2021/yellow_tripdata_2021-03.parquet")
    yellow_tripdata_2021_04 = ProcessYellowFile(spark,f"{yellow_path}/2021/yellow_tripdata_2021-04.parquet")
    yellow_tripdata_2021_05 = ProcessYellowFile(spark,f"{yellow_path}/2021/yellow_tripdata_2021-05.parquet")
    yellow_tripdata_2021_06 = ProcessYellowFile(spark,f"{yellow_path}/2021/yellow_tripdata_2021-06.parquet")
    yellow_tripdata_2021_07 = ProcessYellowFile(spark,f"{yellow_path}/2021/yellow_tripdata_2021-07.parquet")
    yellow_tripdata_2021_08 = ProcessYellowFile(spark,f"{yellow_path}/2021/yellow_tripdata_2021-08.parquet")
    yellow_tripdata_2021_09 = ProcessYellowFile(spark,f"{yellow_path}/2021/yellow_tripdata_2021-09.parquet")
    yellow_tripdata_2021_10 = ProcessYellowFile(spark,f"{yellow_path}/2021/yellow_tripdata_2021-10.parquet")
    yellow_tripdata_2021_11 = ProcessYellowFile(spark,f"{yellow_path}/2021/yellow_tripdata_2021-11.parquet")
    yellow_tripdata_2021_12 = ProcessYellowFile(spark,f"{yellow_path}/2021/yellow_tripdata_2021-12.parquet")

    # 2022
    yellow_tripdata_2022_01 = ProcessYellowFile(spark,f"{yellow_path}/2022/yellow_tripdata_2022-01.parquet")
    yellow_tripdata_2022_02 = ProcessYellowFile(spark,f"{yellow_path}/2022/yellow_tripdata_2022-02.parquet")
    yellow_tripdata_2022_03 = ProcessYellowFile(spark,f"{yellow_path}/2022/yellow_tripdata_2022-03.parquet")
    yellow_tripdata_2022_04 = ProcessYellowFile(spark,f"{yellow_path}/2022/yellow_tripdata_2022-04.parquet")
    yellow_tripdata_2022_05 = ProcessYellowFile(spark,f"{yellow_path}/2022/yellow_tripdata_2022-05.parquet")
    yellow_tripdata_2022_06 = ProcessYellowFile(spark,f"{yellow_path}/2022/yellow_tripdata_2022-06.parquet")
    yellow_tripdata_2022_07 = ProcessYellowFile(spark,f"{yellow_path}/2022/yellow_tripdata_2022-07.parquet")
    yellow_tripdata_2022_08 = ProcessYellowFile(spark,f"{yellow_path}/2022/yellow_tripdata_2022-08.parquet")
    yellow_tripdata_2022_09 = ProcessYellowFile(spark,f"{yellow_path}/2022/yellow_tripdata_2022-09.parquet")
    yellow_tripdata_2022_10 = ProcessYellowFile(spark,f"{yellow_path}/2022/yellow_tripdata_2022-10.parquet")
    yellow_tripdata_2022_11 = ProcessYellowFile(spark,f"{yellow_path}/2022/yellow_tripdata_2022-11.parquet")
    yellow_tripdata_2022_12 = ProcessYellowFile(spark,f"{yellow_path}/2022/yellow_tripdata_2022-12.parquet")

    # 2023
    yellow_tripdata_2023_01 = ProcessYellowFile(spark,f"{yellow_path}/2023/yellow_tripdata_2023-01.parquet")
    yellow_tripdata_2023_02 = ProcessYellowFile(spark,f"{yellow_path}/2023/yellow_tripdata_2023-02.parquet")
    yellow_tripdata_2023_03 = ProcessYellowFile(spark,f"{yellow_path}/2023/yellow_tripdata_2023-03.parquet")
    yellow_tripdata_2023_04 = ProcessYellowFile(spark,f"{yellow_path}/2023/yellow_tripdata_2023-04.parquet")
    yellow_tripdata_2023_05 = ProcessYellowFile(spark,f"{yellow_path}/2023/yellow_tripdata_2023-05.parquet")
    yellow_tripdata_2023_06 = ProcessYellowFile(spark,f"{yellow_path}/2023/yellow_tripdata_2023-06.parquet")
    yellow_tripdata_2023_07 = ProcessYellowFile(spark,f"{yellow_path}/2023/yellow_tripdata_2023-07.parquet")
    yellow_tripdata_2023_08 = ProcessYellowFile(spark,f"{yellow_path}/2023/yellow_tripdata_2023-08.parquet")
    yellow_tripdata_2023_09 = ProcessYellowFile(spark,f"{yellow_path}/2023/yellow_tripdata_2023-09.parquet")
    yellow_tripdata_2023_10 = ProcessYellowFile(spark,f"{yellow_path}/2023/yellow_tripdata_2023-10.parquet")
    yellow_tripdata_2023_11 = ProcessYellowFile(spark,f"{yellow_path}/2023/yellow_tripdata_2023-11.parquet")
    yellow_tripdata_2023_12 = ProcessYellowFile(spark,f"{yellow_path}/2023/yellow_tripdata_2023-12.parquet")

    # 2024
    yellow_tripdata_2024_01 = ProcessYellowFile(spark,f"{yellow_path}/2024/yellow_tripdata_2024-01.parquet")
    yellow_tripdata_2024_02 = ProcessYellowFile(spark,f"{yellow_path}/2024/yellow_tripdata_2024-02.parquet")

    # Merge everything together (we should have all the same schema now)
    df_yellow_final = yellow_tripdata_2019_01 \
        .union(yellow_tripdata_2019_02) \
        .union(yellow_tripdata_2019_03) \
        .union(yellow_tripdata_2019_04) \
        .union(yellow_tripdata_2019_05) \
        .union(yellow_tripdata_2019_06) \
        .union(yellow_tripdata_2019_07) \
        .union(yellow_tripdata_2019_08) \
        .union(yellow_tripdata_2019_09) \
        .union(yellow_tripdata_2019_10) \
        .union(yellow_tripdata_2019_11) \
        .union(yellow_tripdata_2019_12) \
        .union(yellow_tripdata_2020_01) \
        .union(yellow_tripdata_2020_02) \
        .union(yellow_tripdata_2020_03) \
        .union(yellow_tripdata_2020_04) \
        .union(yellow_tripdata_2020_05) \
        .union(yellow_tripdata_2020_06) \
        .union(yellow_tripdata_2020_07) \
        .union(yellow_tripdata_2020_08) \
        .union(yellow_tripdata_2020_09) \
        .union(yellow_tripdata_2020_10) \
        .union(yellow_tripdata_2020_11) \
        .union(yellow_tripdata_2020_12) \
        .union(yellow_tripdata_2021_01) \
        .union(yellow_tripdata_2021_02) \
        .union(yellow_tripdata_2021_03) \
        .union(yellow_tripdata_2021_04) \
        .union(yellow_tripdata_2021_05) \
        .union(yellow_tripdata_2021_06) \
        .union(yellow_tripdata_2021_07) \
        .union(yellow_tripdata_2021_08) \
        .union(yellow_tripdata_2021_09) \
        .union(yellow_tripdata_2021_10) \
        .union(yellow_tripdata_2021_11) \
        .union(yellow_tripdata_2021_12) \
        .union(yellow_tripdata_2022_01) \
        .union(yellow_tripdata_2022_02) \
        .union(yellow_tripdata_2022_03) \
        .union(yellow_tripdata_2022_04) \
        .union(yellow_tripdata_2022_05) \
        .union(yellow_tripdata_2022_06) \
        .union(yellow_tripdata_2022_07) \
        .union(yellow_tripdata_2022_08) \
        .union(yellow_tripdata_2022_09) \
        .union(yellow_tripdata_2022_10) \
        .union(yellow_tripdata_2022_11) \
        .union(yellow_tripdata_2022_12) \
        .union(yellow_tripdata_2023_01) \
        .union(yellow_tripdata_2023_02) \
        .union(yellow_tripdata_2023_03) \
        .union(yellow_tripdata_2023_04) \
        .union(yellow_tripdata_2023_05) \
        .union(yellow_tripdata_2023_06) \
        .union(yellow_tripdata_2023_07) \
        .union(yellow_tripdata_2023_08) \
        .union(yellow_tripdata_2023_09) \
        .union(yellow_tripdata_2023_10) \
        .union(yellow_tripdata_2023_11) \
        .union(yellow_tripdata_2023_12) \
        .union(yellow_tripdata_2024_01) \
        .union(yellow_tripdata_2024_02)

    df_with_partition_cols = df_yellow_final \
        .withColumn("year",  year      (col("Pickup_DateTime"))) \
        .withColumn("month", month     (col("Pickup_DateTime"))) \
        .filter("Pickup_DateTime >= '2019-01-01' AND Pickup_DateTime <= '2024-02-29'")

    # Write as Parquet
    df_with_partition_cols \
        .repartition(5) \
        .coalesce(5) \
        .write \
        .mode("overwrite") \
        .partitionBy("year","month") \
        .parquet(destination + "yellow/trips_table/parquet")

    # Write as CSV
    df_with_partition_cols \
        .repartition(5) \
        .coalesce(5) \
        .write \
        .mode("overwrite") \
        .partitionBy("year","month") \
        .format("csv") \
        .option('header',True) \
        .save(destination + "yellow/trips_table/csv")

    # Write as JSON
    df_with_partition_cols \
        .repartition(5) \
        .coalesce(5) \
        .write \
        .mode("overwrite") \
        .partitionBy("year","month") \
        .format("json") \
        .save(destination + "yellow/trips_table/json")


    ################################################################################################
    # Green
    ################################################################################################    
    green_path = sourceGreen.replace("/*/*.parquet","")
    
   # The NYC taxi data has fields that change data types and some are incompatible with mergeScheme   
    # 2019
    green_tripdata_2019_01 = ProcessGreenFile(spark,f"{green_path}/2019/green_tripdata_2019-01.parquet")
    green_tripdata_2019_02 = ProcessGreenFile(spark,f"{green_path}/2019/green_tripdata_2019-02.parquet")
    green_tripdata_2019_03 = ProcessGreenFile(spark,f"{green_path}/2019/green_tripdata_2019-03.parquet")
    green_tripdata_2019_04 = ProcessGreenFile(spark,f"{green_path}/2019/green_tripdata_2019-04.parquet")
    green_tripdata_2019_05 = ProcessGreenFile(spark,f"{green_path}/2019/green_tripdata_2019-05.parquet")
    green_tripdata_2019_06 = ProcessGreenFile(spark,f"{green_path}/2019/green_tripdata_2019-06.parquet")
    green_tripdata_2019_07 = ProcessGreenFile(spark,f"{green_path}/2019/green_tripdata_2019-07.parquet")
    green_tripdata_2019_08 = ProcessGreenFile(spark,f"{green_path}/2019/green_tripdata_2019-08.parquet")
    green_tripdata_2019_09 = ProcessGreenFile(spark,f"{green_path}/2019/green_tripdata_2019-09.parquet")
    green_tripdata_2019_10 = ProcessGreenFile(spark,f"{green_path}/2019/green_tripdata_2019-10.parquet")
    green_tripdata_2019_11 = ProcessGreenFile(spark,f"{green_path}/2019/green_tripdata_2019-11.parquet")
    green_tripdata_2019_12 = ProcessGreenFile(spark,f"{green_path}/2019/green_tripdata_2019-12.parquet")

    # 2020
    green_tripdata_2020_01 = ProcessGreenFile(spark,f"{green_path}/2020/green_tripdata_2020-01.parquet")
    green_tripdata_2020_02 = ProcessGreenFile(spark,f"{green_path}/2020/green_tripdata_2020-02.parquet")
    green_tripdata_2020_03 = ProcessGreenFile(spark,f"{green_path}/2020/green_tripdata_2020-03.parquet")
    green_tripdata_2020_04 = ProcessGreenFile(spark,f"{green_path}/2020/green_tripdata_2020-04.parquet")
    green_tripdata_2020_05 = ProcessGreenFile(spark,f"{green_path}/2020/green_tripdata_2020-05.parquet")
    green_tripdata_2020_06 = ProcessGreenFile(spark,f"{green_path}/2020/green_tripdata_2020-06.parquet")
    green_tripdata_2020_07 = ProcessGreenFile(spark,f"{green_path}/2020/green_tripdata_2020-07.parquet")
    green_tripdata_2020_08 = ProcessGreenFile(spark,f"{green_path}/2020/green_tripdata_2020-08.parquet")
    green_tripdata_2020_09 = ProcessGreenFile(spark,f"{green_path}/2020/green_tripdata_2020-09.parquet")
    green_tripdata_2020_10 = ProcessGreenFile(spark,f"{green_path}/2020/green_tripdata_2020-10.parquet")
    green_tripdata_2020_11 = ProcessGreenFile(spark,f"{green_path}/2020/green_tripdata_2020-11.parquet")
    green_tripdata_2020_12 = ProcessGreenFile(spark,f"{green_path}/2020/green_tripdata_2020-12.parquet")

    # 2021
    green_tripdata_2021_01 = ProcessGreenFile(spark,f"{green_path}/2021/green_tripdata_2021-01.parquet")
    green_tripdata_2021_02 = ProcessGreenFile(spark,f"{green_path}/2021/green_tripdata_2021-02.parquet")
    green_tripdata_2021_03 = ProcessGreenFile(spark,f"{green_path}/2021/green_tripdata_2021-03.parquet")
    green_tripdata_2021_04 = ProcessGreenFile(spark,f"{green_path}/2021/green_tripdata_2021-04.parquet")
    green_tripdata_2021_05 = ProcessGreenFile(spark,f"{green_path}/2021/green_tripdata_2021-05.parquet")
    green_tripdata_2021_06 = ProcessGreenFile(spark,f"{green_path}/2021/green_tripdata_2021-06.parquet")
    green_tripdata_2021_07 = ProcessGreenFile(spark,f"{green_path}/2021/green_tripdata_2021-07.parquet")
    green_tripdata_2021_08 = ProcessGreenFile(spark,f"{green_path}/2021/green_tripdata_2021-08.parquet")
    green_tripdata_2021_09 = ProcessGreenFile(spark,f"{green_path}/2021/green_tripdata_2021-09.parquet")
    green_tripdata_2021_10 = ProcessGreenFile(spark,f"{green_path}/2021/green_tripdata_2021-10.parquet")
    green_tripdata_2021_11 = ProcessGreenFile(spark,f"{green_path}/2021/green_tripdata_2021-11.parquet")
    green_tripdata_2021_12 = ProcessGreenFile(spark,f"{green_path}/2021/green_tripdata_2021-12.parquet")

    # 2022
    green_tripdata_2022_01 = ProcessGreenFile(spark,f"{green_path}/2022/green_tripdata_2022-01.parquet")
    green_tripdata_2022_02 = ProcessGreenFile(spark,f"{green_path}/2022/green_tripdata_2022-02.parquet")
    green_tripdata_2022_03 = ProcessGreenFile(spark,f"{green_path}/2022/green_tripdata_2022-03.parquet")
    green_tripdata_2022_04 = ProcessGreenFile(spark,f"{green_path}/2022/green_tripdata_2022-04.parquet")
    green_tripdata_2022_05 = ProcessGreenFile(spark,f"{green_path}/2022/green_tripdata_2022-05.parquet")
    green_tripdata_2022_06 = ProcessGreenFile(spark,f"{green_path}/2022/green_tripdata_2022-06.parquet")
    green_tripdata_2022_07 = ProcessGreenFile(spark,f"{green_path}/2022/green_tripdata_2022-07.parquet")
    green_tripdata_2022_08 = ProcessGreenFile(spark,f"{green_path}/2022/green_tripdata_2022-08.parquet")
    green_tripdata_2022_09 = ProcessGreenFile(spark,f"{green_path}/2022/green_tripdata_2022-09.parquet")
    green_tripdata_2022_10 = ProcessGreenFile(spark,f"{green_path}/2022/green_tripdata_2022-10.parquet")
    green_tripdata_2022_11 = ProcessGreenFile(spark,f"{green_path}/2022/green_tripdata_2022-11.parquet")
    green_tripdata_2022_12 = ProcessGreenFile(spark,f"{green_path}/2022/green_tripdata_2022-12.parquet")

    # 2023
    green_tripdata_2023_01 = ProcessGreenFile(spark,f"{green_path}/2023/green_tripdata_2023-01.parquet")
    green_tripdata_2023_02 = ProcessGreenFile(spark,f"{green_path}/2023/green_tripdata_2023-02.parquet")
    green_tripdata_2023_03 = ProcessGreenFile(spark,f"{green_path}/2023/green_tripdata_2023-03.parquet")
    green_tripdata_2023_04 = ProcessGreenFile(spark,f"{green_path}/2023/green_tripdata_2023-04.parquet")
    green_tripdata_2023_05 = ProcessGreenFile(spark,f"{green_path}/2023/green_tripdata_2023-05.parquet")
    green_tripdata_2023_06 = ProcessGreenFile(spark,f"{green_path}/2023/green_tripdata_2023-06.parquet")
    green_tripdata_2023_07 = ProcessGreenFile(spark,f"{green_path}/2023/green_tripdata_2023-07.parquet")
    green_tripdata_2023_08 = ProcessGreenFile(spark,f"{green_path}/2023/green_tripdata_2023-08.parquet")
    green_tripdata_2023_09 = ProcessGreenFile(spark,f"{green_path}/2023/green_tripdata_2023-09.parquet")
    green_tripdata_2023_10 = ProcessGreenFile(spark,f"{green_path}/2023/green_tripdata_2023-10.parquet")
    green_tripdata_2023_11 = ProcessGreenFile(spark,f"{green_path}/2023/green_tripdata_2023-11.parquet")
    green_tripdata_2023_12 = ProcessGreenFile(spark,f"{green_path}/2023/green_tripdata_2023-12.parquet")

    # 2024
    green_tripdata_2024_01 = ProcessGreenFile(spark,f"{green_path}/2024/green_tripdata_2024-01.parquet")
    green_tripdata_2024_02 = ProcessGreenFile(spark,f"{green_path}/2024/green_tripdata_2024-02.parquet")

    # Merge everything together (we should have all the same schema now)
    df_green_final = green_tripdata_2019_01 \
        .union(green_tripdata_2019_02) \
        .union(green_tripdata_2019_03) \
        .union(green_tripdata_2019_04) \
        .union(green_tripdata_2019_05) \
        .union(green_tripdata_2019_06) \
        .union(green_tripdata_2019_07) \
        .union(green_tripdata_2019_08) \
        .union(green_tripdata_2019_09) \
        .union(green_tripdata_2019_10) \
        .union(green_tripdata_2019_11) \
        .union(green_tripdata_2019_12) \
        .union(green_tripdata_2020_01) \
        .union(green_tripdata_2020_02) \
        .union(green_tripdata_2020_03) \
        .union(green_tripdata_2020_04) \
        .union(green_tripdata_2020_05) \
        .union(green_tripdata_2020_06) \
        .union(green_tripdata_2020_07) \
        .union(green_tripdata_2020_08) \
        .union(green_tripdata_2020_09) \
        .union(green_tripdata_2020_10) \
        .union(green_tripdata_2020_11) \
        .union(green_tripdata_2020_12) \
        .union(green_tripdata_2021_01) \
        .union(green_tripdata_2021_02) \
        .union(green_tripdata_2021_03) \
        .union(green_tripdata_2021_04) \
        .union(green_tripdata_2021_05) \
        .union(green_tripdata_2021_06) \
        .union(green_tripdata_2021_07) \
        .union(green_tripdata_2021_08) \
        .union(green_tripdata_2021_09) \
        .union(green_tripdata_2021_10) \
        .union(green_tripdata_2021_11) \
        .union(green_tripdata_2021_12) \
        .union(green_tripdata_2022_01) \
        .union(green_tripdata_2022_02) \
        .union(green_tripdata_2022_03) \
        .union(green_tripdata_2022_04) \
        .union(green_tripdata_2022_05) \
        .union(green_tripdata_2022_06) \
        .union(green_tripdata_2022_07) \
        .union(green_tripdata_2022_08) \
        .union(green_tripdata_2022_09) \
        .union(green_tripdata_2022_10) \
        .union(green_tripdata_2022_11) \
        .union(green_tripdata_2022_12) \
        .union(green_tripdata_2023_01) \
        .union(green_tripdata_2023_02) \
        .union(green_tripdata_2023_03) \
        .union(green_tripdata_2023_04) \
        .union(green_tripdata_2023_05) \
        .union(green_tripdata_2023_06) \
        .union(green_tripdata_2023_07) \
        .union(green_tripdata_2023_08) \
        .union(green_tripdata_2023_09) \
        .union(green_tripdata_2023_10) \
        .union(green_tripdata_2023_11) \
        .union(green_tripdata_2023_12) \
        .union(green_tripdata_2024_01) \
        .union(green_tripdata_2024_02)

    df_with_partition_cols = df_green_final \
        .withColumn("year",  year      (col("Pickup_DateTime"))) \
        .withColumn("month", month     (col("Pickup_DateTime"))) \
        .filter("Pickup_DateTime >= '2019-01-01' AND Pickup_DateTime <= '2024-02-29'")

    # Write as Parquet
    df_with_partition_cols \
        .repartition(5) \
        .coalesce(5) \
        .write \
        .mode("overwrite") \
        .partitionBy("year","month") \
        .parquet(destination + "green/trips_table/parquet")

    # Write as CSV
    df_with_partition_cols \
        .repartition(5) \
        .coalesce(5) \
        .write \
        .mode("overwrite") \
        .partitionBy("year","month") \
        .format("csv") \
        .option('header',True) \
        .save(destination + "green/trips_table/csv")

    # Write as JSON
    df_with_partition_cols \
        .repartition(5) \
        .coalesce(5) \
        .write \
        .mode("overwrite") \
        .partitionBy("year","month") \
        .format("json") \
        .save(destination + "green/trips_table/json")
    

    ################################################################################################
    # Create Common Tables 
    ################################################################################################
    # Vendor Table
    vendor_df = spark.createDataFrame(
        [
            (1, "Creative Mobile Technologies"), 
            (2, "VeriFone")
        ],
        ["Vendor_Id", "Vendor_Description"] 
    )        

    vendor_df \
        .repartition(1) \
        .coalesce(1) \
        .write \
        .mode("overwrite") \
        .parquet(destination + "vendor_table/")

    # Rate Code Table
    rate_code_df = spark.createDataFrame(
        [
            (1, "Standard rate"), 
            (2, "JFK"),
            (3, "Newark"),
            (4, "Nassau or Westchester"),
            (5, "Negotiated fare"),
            (6, "Group ride"),
        ],
        ["Rate_Code_Id", "Rate_Code_Description"] 
    )        

    rate_code_df \
        .repartition(1) \
        .coalesce(1) \
        .write \
        .mode("overwrite") \
        .parquet(destination + "rate_code_table/")

    # Payment Type Table
    payment_type_df = spark.createDataFrame(
        [
            (1, "Credit card"), 
            (2, "Cash"),
            (3, "No charge"),
            (4, "Dispute"),
            (5, "Unknown"),
            (6, "Voided trip"),
        ],
        ["Payment_Type_Id", "Payment_Type_Description"] 
    )        

    payment_type_df \
        .repartition(1) \
        .coalesce(1) \
        .write \
        .mode("overwrite") \
        .parquet(destination + "payment_type_table/")

    # Trip Type Table
    trip_type_df = spark.createDataFrame(
        [
            (1, "Street-hail"),
            (2, "Dispatch")
        ],
        ["Trip_Type_Id", "Trip_Type_Description"] 
    )        

    trip_type_df \
        .repartition(1) \
        .coalesce(1) \
        .write \
        .mode("overwrite") \
        .parquet(destination + "trip_type_table/")

    spark.stop()


# Main entry point
# convert_taxi_to_parquet gs://big-query-demo-09/test-taxi/yellow gs://big-query-demo-09/test-taxi/green gs://big-query-demo-09/test-taxi-output
if __name__ == "__main__":
    if len(sys.argv) != 4:
        print("Usage: convert_taxi_to_parquet sourceYellow sourceGreen destination")
        sys.exit(-1)

    sourceYellow = sys.argv[1]
    sourceGreen  = sys.argv[2]
    destination  = sys.argv[3]

    print ("BEGIN: Main")
    ConvertTaxiData(sourceYellow, sourceGreen, destination)
    print ("END: Main")

# Sample run 
# gcloud dataproc jobs submit pyspark  \
#    --cluster "testcluster" \
#    --region="REPLACE-REGION" \
#    gs://big-query-demo-09/pyspark-code/convert_taxi_to_parquet.py \
#    -- gs://big-query-demo-09/test-taxi/yellow/*/*.parquet \
#       gs://big-query-demo-09/test-taxi/green/*/*.parquet \
#       gs://big-query-demo-09/test-taxi/dest/