Untitled

import pyspark.sql.functions as F
from pyspark.sql import SparkSession
from pyspark.sql.types import StructType, StructField, StringType, IntegerType, DoubleType, DateType

# =============================================================================
# SETUP: Initialize Spark and Create 5 Sample DataFrames
# =============================================================================

spark = SparkSession.builder.appName("ToyotaDEPractice_5_Tables").getOrCreate()

# -- Table 1: Dealers (Small Dimension Table) --
dealer_data = [
    (1, "Toyota of Bellevue", "Bellevue", "wa"), # Inconsistent state code
    (2, "Seattle Toyota", "Seattle", "WA"),
    (3, "Portland Auto", "Portland", "OR"),
    (4, "LA Cars", "Los Angeles", "CA"),
]
dealer_schema = StructType([
    StructField("dealer_id", IntegerType(), True),
    StructField("dealer_name", StringType(), True),
    StructField("city", StringType(), True),
    StructField("state", StringType(), True),
])
dealers_df = spark.createDataFrame(data=dealer_data, schema=dealer_schema)

# -- Table 2: Customers (Dimension Table) --
customer_data = [
    (101, "John Smith", "2022-01-15"),
    (102, "Jane Doe", "2022-05-20"),
    (103, "Peter Jones", "2023-03-10"),
    (104, "Mary Garcia", "2023-07-22"),
    (105, "David Miller", "2024-02-01"),
]
customer_schema = StructType([
    StructField("customer_id", IntegerType(), True),
    StructField("customer_name", StringType(), True),
    StructField("signup_date", StringType(), True),
])
customers_df = spark.createDataFrame(data=customer_data, schema=customer_schema)
customers_df = customers_df.withColumn("signup_date", F.to_date(F.col("signup_date")))


# -- Table 3: Vehicles (Dimension Table) --
vehicle_data = [
    ("VIN001", "Camry", 2023, "Hybrid"),
    ("VIN002", "RAV4", 2023, "Gas"),
    ("VIN003", "Tacoma", 2024, "Gas"),
    ("VIN004", "Corolla", 2023, "Hybrid"),
    ("VIN005", "Highlander", 2024, "Hybrid"),
    ("VIN006", "RAV4", 2024, "Gas"),
]
vehicle_schema = StructType([
    StructField("vin", StringType(), True),
    StructField("model", StringType(), True),
    StructField("year", IntegerType(), True),
    StructField("engine_type", StringType(), True),
])
vehicles_df = spark.createDataFrame(data=vehicle_data, schema=vehicle_schema)

# -- Table 4: Sales (Fact Table) --
sales_data = [
    (1001, "VIN001", 1, 101, "2023-03-15", 35000.0),
    (1002, "VIN002", 2, 102, "2023-04-01", 38000.0),
    (1003, "VIN004", 1, 103, "2023-05-20", 28000.0),
    (1004, "VIN003", 3, 101, "2024-01-10", 45000.0),
    (1005, "VIN005", 4, 104, "2024-03-01", 52000.0),
    (1006, "VIN001", 1, 102, "2023-08-10", 34500.0), # Same car, resold
    (1007, "VIN006", 2, 105, "2024-04-05", None), # Null price
]
sales_schema = StructType([
    StructField("sale_id", IntegerType(), True),
    StructField("vin", StringType(), True),
    StructField("dealer_id", IntegerType(), True),
    StructField("customer_id", IntegerType(), True),
    StructField("sale_date", StringType(), True),
    StructField("sale_price", DoubleType(), True),
])
sales_df = spark.createDataFrame(data=sales_data, schema=sales_schema)
sales_df = sales_df.withColumn("sale_date", F.to_date(F.col("sale_date")))

# -- Table 5: Service Logs (Fact Table) --
service_log_data = [
    (2001, "VIN001", "2023-09-12", 5010.0, "Oil Change"),
    (2002, "VIN002", "2023-10-01", 5500.0, "Tire Rotation"),
    (2003, "VIN001", "2024-03-15", 10050.0, "Brake Inspection"),
    (2004, "VIN004", "2023-11-25", 6200.0, "Oil Change"),
    (2005, "VIN003", "2023-12-20", 1200.0, "Recall Service"), # Note: Service before sale
    (2006, "VIN005", "2024-08-30", 5150.0, "Oil Change"),
]
service_schema = StructType([
    StructField("log_id", IntegerType(), True),
    StructField("vin", StringType(), True),
    StructField("service_date", StringType(), True),
    StructField("mileage", DoubleType(), True),
    StructField("service_type", StringType(), True),
])
service_logs_df = spark.createDataFrame(data=service_log_data, schema=service_schema)
service_logs_df = service_logs_df.withColumn("service_date", F.to_date(F.col("service_date")))


# =============================================================================
# QUESTIONS FOR YOUR 45-MINUTE PRACTICE
# =============================================================================

# 1. Multi-Table Data Cleaning
# Task: Prepare the data for analysis by performing the following cleaning steps:
#   a. In `dealers_df`, standardize the `state` column to be uppercase.
#   b. In `sales_df`, fill the null `sale_price` with the average sale price for that specific dealer (`dealer_id`).

# Your Answer Here:


# 2. Multi-Join Aggregation
# Task: Calculate the total sales revenue generated in each `state`. The final output should have two columns: `state` and `total_revenue`.

# Your Answer Here:


# 3. Left Join to Find Inactive Customers
# Task: Identify customers who have signed up but have never purchased a vehicle.
# Display the `customer_id` and `customer_name`.

# Your Answer Here:


# 4. Join Strategy (Conceptual)
# Task: When joining `sales_df` (a large fact table) with `dealers_df` (a small dimension table),
# what join strategy should you use to optimize performance and why? Write the code for Question #2 again using this optimal strategy.

# Your Answer Here:


# 5. Ranking with Window Functions
# Task: Within each `state`, rank dealers by their total sales volume (number of cars sold).
# The output should include `state`, `dealer_name`, `cars_sold`, and `rank`.

# Your Answer Here:


# 6. Time-Series Analysis with Window Functions (Hard)
# Task: For each `vin`, find the number of days between its first sale and its first service appointment.
# The output should have `vin`, `model`, and `days_to_first_service`.

# Your Answer Here:


# 7. Data Quality & Integrity Check (Complex Logic)
# Task: Identify any service records that occurred *before* the vehicle's first sale date.
# Such a record indicates a data quality issue. Display the full service log row for any such records.

# Your Answer Here:


# 8. Set Operations
# Task: Create a single DataFrame with a list of all unique `customer_id`s who either purchased a "Hybrid" vehicle OR purchased any vehicle from a dealer in "WA".

# Your Answer Here:


# 9. Pivoting for Reporting
# Task: Create a report that shows the number of vehicles sold by each `dealer_name` for each `year`.
# The final DataFrame should have dealer names as rows and years as columns.

# Your Answer Here:


# 10. Multi-Step Complex Query (Hard)
# Task: Identify the top 2 customers based on their total spending. For these top customers,
# list their names and the total number of service appointments their vehicles have had.

# Your Answer Here: