Spark on AWS

from pyspark.sql import SparkSession
from pyspark.sql.functions import col, to_date, year, month
from delta.tables import DeltaTable

# 1. Create SparkSession with S3 and Delta Lake support
spark = SparkSession.builder \
    .appName("SalesETL_S3") \
    .config("spark.sql.extensions",
            "io.delta.sql.DeltaSparkSessionExtension") \
    .config("spark.sql.catalog.spark_catalog",
            "org.apache.spark.sql.delta.catalog.DeltaCatalog") \
    .getOrCreate()

# S3 bucket paths (replace with your bucket)
RAW_PATH    = "s3://my-company-datalake/raw/sales/"
SILVER_PATH = "s3://my-company-datalake/silver/sales_clean/"

# 2. Read raw JSON from S3 (Bronze layer)
raw_df = spark.read \
    .option("multiLine", True) \
    .json(RAW_PATH)

print(f"Loaded {raw_df.count():,} raw records")

# 3. Clean and transform (Silver layer)
clean_df = raw_df \
    .filter(col("sale_amount") > 0) \
    .filter(col("customer_id").isNotNull()) \
    .withColumn("sale_date", to_date(col("sale_ts"))) \
    .withColumn("year",  year(col("sale_date"))) \
    .withColumn("month", month(col("sale_date"))) \
    .select(
        "customer_id", "product_id", "sale_date",
        "sale_amount", "year", "month"
    )

# 4. Write to S3 as partitioned Delta table (Silver layer)
clean_df.write \
    .format("delta") \
    .mode("append") \
    .partitionBy("year", "month") \
    .save(SILVER_PATH)

print(f"Wrote {clean_df.count():,} clean records to Delta on S3")

# 5. Optimise the Delta table (run periodically)
dt = DeltaTable.forPath(spark, SILVER_PATH)
spark.sql(f"""
    OPTIMIZE delta.`${SILVER_PATH}`
    ZORDER BY (customer_id, sale_date)
""")

# The Delta table is now queryable via Amazon Athena with:
# SELECT * FROM delta_lake_table WHERE year=2024 AND month=3