How do I actually set up the SparkSession and the entry point for a production batch job?

When you're moving from a notebook to a real Scala application, the biggest shift is how you manage the SparkSession. You don't want to be manually creating sessions all over your code. I usually wrap the logic in a main object and keep the session configuration centralized. In a real production environment, you won't hardcode the config; you'll let spark-submit handle the master and memory settings, but you still need the builder in your code to kick things off.

import org.apache.spark.sql.SparkSession

object SalesETL {
  def main(args: Array[String]): Unit = {
    val spark = SparkSession.builder()
      .appName("DailySalesAggregation")
      .getOrCreate()

    try {
      // Your ETL logic goes here
      val job = new SalesProcessor(spark)
      job.run()
    } finally {
      spark.stop()
    }
  }
}

Pro tip: Always wrap your execution in a try-finally block to ensure the session stops. It sounds trivial, but in shared cluster environments, orphaned sessions are a great way to get a stern email from your DevOps team.

How do I handle the "T" in ETL without creating a giant, unreadable chain of transformations?
This is where most developers trip up. They write one massive 50-line chain of .filter().select().withColumn() and then spend three days debugging a null pointer exception. Instead, I treat my transformations as a series of small, testable functions. Since we're in Scala, we can leverage case classes to give our data structure and type safety, moving from a generic DataFrame to a typed Dataset.
Let's say we're processing raw JSON transaction logs. I'd split the cleaning from the aggregation:
case class RawTransaction(transactionId: String, customerId: String, amount: Double, timestamp: String)
case class CleanTransaction(transactionId: String, customerId: String, amount: Double, date: java.sql.Date)

class SalesProcessor(spark: SparkSession) {
  import spark.implicits._

  def run(): Unit = {
    val rawData = spark.read.json("s3://bucket/raw/transactions/*.json").as[RawTransaction]
    
    val cleanedData = cleanTransactions(rawData)
    val dailySummary = aggregateSales(cleanedData)
    
    saveResults(dailySummary)
  }

  def cleanTransactions(ds: Dataset[RawTransaction]): Dataset[CleanTransaction] = {
    ds.filter(_.amount > 0) // Remove errors or returns
      .map(tx => CleanTransaction(
        tx.transactionId, 
        tx.customerId, 
        tx.amount, 
        java.sql.Date.valueOf(tx.timestamp.substring(0, 10))
      ))
  }

  def aggregateSales(ds: Dataset[CleanTransaction]) = {
    ds.groupBy("date")
      .sum("amount")
      .withColumnRenamed("sum(amount)", "total_revenue")
  }
}


What's the best way to write the data so it's actually performant for the people using it?
Writing a single massive CSV file is the fastest way to make your data analysts hate you. In a batch ETL job, you want to write in a columnar format—Parquet is the gold standard here—and you should almost always partition your data. Partitioning physically organizes the data into folders on disk, which means Spark (and other tools like Athena or Presto) can skip entire directories of data when querying a specific date.
I usually partition by the coarsest grain needed for reporting, like year, month, or date. Here is how I'd handle the write step for our sales job:
def saveResults(df: org.apache.spark.sql.DataFrame): Unit = {
  df.write
    .mode("overwrite")
    .partitionBy("date")
    .parquet("s3://bucket/gold/daily_sales_summary/")
}

By using partitionBy("date"), Spark creates a folder structure like date=2023-10-01/. When someone queries for just one day, they aren't scanning the entire dataset. It's a massive win for both cost and speed.

📋 Practical Task

Exercise: Building a Customer Lifetime Value (CLV) Batch Aggregator

Your goal is to build a batch ETL job that calculates the total spend per customer across multiple transaction files. You will be provided with a directory of JSON files containing transactions with the following schema: { "customerId": String, "amount": Double, "transactionDate": String }.

Requirements:

• Create a case class to represent the transaction.
• Implement a transformation pipeline that:
  1. Filters out any transactions where the amount is negative.
  2. Groups the data by customerId.
  3. Calculates the total sum of amount for each customer.
• Write the resulting Dataset to a Parquet file located at /tmp/customer_clv/.
• Ensure the final output is partitioned by a newly created column called customer_segment (for this exercise, you can hardcode this as "General" using withColumn).

Evaluation Criteria: Your solution should demonstrate the separation of the SparkSession setup from the transformation logic and use a typed Dataset for the initial cleaning phase.