28 Jun 2025

Functional utilities for data engineering: stop repeats in Scala pipelines now

You’re writing a Spark pipeline on GCP. Again. You need to:

Handle nulls and parse dates
Wrap risky operations in try-catch
Find affected partitions since last run (which strategy? GCS multi-threaded? DataFrame-based? Bucket-level?)
Work with GCS blobs using clean syntax (blob"gs://bucket/file")
Vacuum your Delta tables, handle schema evolution, run data quality checks

You copy-paste from last week’s code. The pipeline works, but you know this pattern will show up ten more times this quarter.

“Can I just… not repeat myself? Like, ever again?”

I had this exact frustration after writing my 50th affected partitions check and manually running OPTIMIZE commands on every Delta table. So I built a comprehensive library of functional utilities that turned repetitive pipeline patterns into composable, type-safe operations.

This post shows you how to build these utilities from scratch - not just what they do, but WHY they exist, HOW they work internally, and WHEN to use them. We’ll cover:

Higher-order functions that wrap unsafe operations
Implicit classes that extend types without inheritance
String interpolators for dates AND GCS blobs
5 affected partitions strategies (GCS multi-threaded, Spark DataFrame-based, bucket-level, file-level, view-level with filters)
GCS operations (blob/blobs/bucket interpolators, copy/move/timestamps, parallel operations)
Real CDC (Change Data Capture) with SCD Type 1
Delta Lake utilities (vacuum, compaction, Z-ordering)
Schema evolution strategies (partitioned vs non-partitioned)
Data quality wrappers that integrate DQ frameworks
Schema comparison utilities for detecting changes
Production backup strategies that work across Spark

All code runs on Scala 2.13 + Spark 3.x + Delta Lake + GCP. You can grab the working implementation from dataengineering-savvy on GitHub .

Information

The code in this post is battle-tested in production data pipelines processing TBs daily. Each utility solved a real pain point.

Why functional utilities matter

Here’s the thing: data engineering isn’t just about moving data. It’s about doing it safely, repeatedly, and without surprises.

When you’re building pipelines:

Operations fail (null pointers, network timeouts, bad data)
Dates come in 15 different formats
You need CDC logic that works the same way every time
Delta tables need maintenance (vacuum, compaction, Z-ordering)
Schemas evolve - columns get added/dropped
Data quality checks must run consistently
Table backups can’t just be “copy files” - partitions matter
Type conversions should be safe, not crashy

Writing this logic inline every time leads to:

Inconsistent error handling - Some devs throw, some return Option, some log and swallow
Copy-paste bugs - That date parser you copied? It has a timezone bug
Unreadable code - LocalDate.parse(dateStr, DateTimeFormatter.ofPattern("yyyy-MM-dd")) vs yearMonthDate"2025-06-28"
No reuse - The CDC logic you wrote for Customer? Can’t reuse it for Orders without modification
Manual ops - Running OPTIMIZE and VACUUM on every Delta table by hand

Functional utilities fix this by:

Wrapping common patterns in higher-order functions (functions that take functions as arguments)
Using Scala’s type system to prevent runtime errors at compile time
Making code read like English through well-designed APIs
Enforcing consistency - one way to do CDC, one way to handle errors, one way to parse dates, one way to optimize Delta tables
Automating repetitive ops - schema evolution, data quality, table maintenance

The compiler becomes your ally. If it compiles, it handles errors correctly.

Production gotcha

Don’t confuse compile-time type safety with runtime data validation. These utilities prevent structure errors (wrong types, missing methods), not data errors (null IDs, negative amounts, malformed JSON). You still need data quality checks - the utilities just make implementing those checks safer and more consistent.

graph TB
    A[Repetitive Pipeline Code] -->|Functional Utilities| B[Composable Abstractions]
    B --> C[Higher-Order Functions]
    B --> D[Implicit Classes]
    B --> E[String Interpolators]
    B --> F[Type Safety]

    C --> G[Error Handling]
    D --> H[Type Extensions]
    E --> I[Domain-Specific Syntax]
    F --> J[Compile-Time Guarantees]

    G --> K[Production Pipelines]
    H --> K
    I --> K
    J --> K

    style K fill:#90EE90,stroke:#2d7a2d,color:#000
    style A fill:#FFB6C6,stroke:#cc0066,color:#000

What we’ll build

We’re going to build eleven categories of utilities:

Error handling - trySafely: Higher-order wrapper for unsafe operations
Type conversions - Java ↔ Scala, JSON ↔ Case classes
Date/time interpolators - yearMonthDate"2025-06-28" instead of parsing boilerplate
GCS string interpolators - blob"gs://bucket/file" for clean cloud storage access
Spark CDC - Change Data Capture with SCD Type 1 logic
Delta Lake ops - Vacuum, compaction, Z-ordering wrappers
Schema evolution - Automatic column add/drop with validation
Data quality - Implicit wrappers for DQ frameworks
Schema comparison - Detect schema changes between DataFrames
Table operations - Backup, partition management, location discovery
GCS object operations - Copy, move, timestamps, parallel operations for cloud files
Affected partitions - 5 strategies for finding what changed (DataFrame-based, GCS multi-threaded, bucket-level, file-level, view-level)

Each utility uses a different functional pattern. By the end, you’ll understand:

How implicit classes extend types
How higher-order functions capture behavior
How string interpolators work under the hood
How to design APIs that prevent misuse
How to wrap complex operations in simple interfaces

Let’s start with the foundation: safe error handling.

Part 1 - Higher-order functions: Wrapping unsafe operations

WHY we need this

Every data pipeline has unsafe operations:

1
2
3
4
5
6
7
8
// Division by zero
val result = totalRevenue / orderCount  // Crashes if orderCount = 0

// API calls
val response = httpClient.get(url)  // Throws if network fails

// File operations
val content = Source.fromFile("data.csv").getLines()  // Throws FileNotFoundException

The naive approach:

1
2
3
4
5
6
7
8
9
try {
  val result = riskyOperation()
  result
} catch {
  case ex: Exception =>
    logger.error("Something went wrong")
    ex.printStackTrace()
    -1  // Or null, or Option.empty, or throw again...
}

Problems:

Every developer handles errors differently
Copy-pasted everywhere
Hard to test
Inconsistent fallback strategies

We need one function that handles all try-catch logic, makes error handling explicit, and is fully generic.

HOW it works: Higher-order functions

A higher-order function is a function that:

Takes functions as parameters
Returns functions as results
Or both

Think of it like a template: “Give me an unsafe operation, an error message, and a fallback - I’ll handle the try-catch for you.”

Here’s the signature:

1
2
3
4
5
def trySafely[T, E <: Throwable, C](
  unsafeCodeBlock: => T,                     // The risky operation
  errorMessage: Option[String],              // Message to log
  exceptionHandlingCodeblock: => Option[C]   // Fallback behavior
): Either[C, T]

What’s happening here:

unsafeCodeBlock: => T - The => means “call by name” (lazy evaluation). We don’t execute this immediately - we pass it AS CODE to the function, and it executes inside the try-catch.
[T, E <: Throwable, C] - Three generic types:
- T: Return type of the unsafe operation
- E: Exception type (constrained to Throwable hierarchy)
- C: Return type of the fallback handler
Either[C, T] - Returns Left(fallback) on failure, Right(result) on success. This forces the caller to handle both cases.

flowchart LR
    A[Unsafe Operation] --> B{trySafely}
    B -->|Try| C[Execute Code Block]
    C -->|Success| D[Right result]
    C -->|Failure| E[handleException]
    E -->|Has Fallback| F[Left fallback]
    E -->|No Fallback| G[Throw Exception]

    style D fill:#90EE90,stroke:#2d7a2d,color:#000
    style F fill:#FFD700,stroke:#cc8800,color:#000
    style G fill:#FFB6C6,stroke:#cc0066,color:#000

WHAT the implementation looks like

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
def trySafely[T, E <: Throwable, C](
  unsafeCodeBlock: => T,
  errorMessage: Option[String],
  exceptionHandlingCodeblock: => Option[C] = None
): Either[C, T] = {
  val message: String = errorMessage.getOrElse("ERROR: Operation failed")

  Try {
    unsafeCodeBlock  // Execute the risky operation
  } match {
    case Success(codeBlockResult) =>
      Right(codeBlockResult)  // Success path
    case Failure(exception: Throwable) =>
      Left(
        handleException(
          exception = exception,
          errorMessage = message,
          exceptionHandlingCodeblock = exceptionHandlingCodeblock
        )
      )
  }
}

private def handleException[E <: Throwable, C](
  exception: E,
  errorMessage: String,
  exceptionHandlingCodeblock: => Option[C]
): C = {
  val codeBlockResult = exceptionHandlingCodeblock.map(c => c)

  if (codeBlockResult.isEmpty) {
    logger.error(errorMessage)
    exception.printStackTrace()
    throw exception  // Re-throw if no fallback
  }

  codeBlockResult.get  // Return fallback result
}

Step-by-step:

Wrap the unsafe code in Scala’s Try { ... }
Pattern match on Success/Failure
On success: Return Right(result) - the happy path
On failure: Call handleException which either:
- Executes the fallback code and returns Left(fallback)
- Logs the error and re-throws if no fallback provided

Usage examples

Example 1: Division with fallback

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
import com.vitthalmirji.dataengineering.common.Helpers.trySafely

val result = trySafely(
  unsafeCodeBlock = 100 / 0,  // This will throw ArithmeticException
  errorMessage = Some("Cannot divide by zero"),
  exceptionHandlingCodeblock = Some(-999)  // Fallback value
)

result match {
  case Right(value) => println(s"Success: $value")
  case Left(fallback) => println(s"Failed, using fallback: $fallback")  // Prints -999
}

Example 2: Spark transformation with logging

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
val transformedDf = trySafely(
  unsafeCodeBlock = {
    sourceDf
      .filter($"amount" > 0)
      .groupBy($"customer_id")
      .agg(sum($"amount"))
  },
  errorMessage = Some("Transformation failed - check schema"),
  exceptionHandlingCodeblock = Some(spark.emptyDataFrame)  // Return empty DF on failure
) match {
  case Right(df) => df
  case Left(emptyDf) => emptyDf
}

Why this pattern works

Type-safe: The compiler ensures you handle both success and failure
Reusable: Write once, use everywhere
Testable: Easy to mock and verify error paths
Consistent: Everyone uses the same error handling strategy
Explicit: Either[C, T] forces you to acknowledge failure is possible

This is the foundation. Every other utility builds on this pattern: “Wrap complexity in a function, make it generic, return something that forces the caller to think.”

Part 2 - Implicit classes: Extending types without inheritance

WHY we need implicit conversions

Imagine you have Java types (java.util.List, java.util.Optional) but you want Scala collections (List, Option). You could write conversion functions:

1
2
3
4
5
6
def javaListToScalaList[T](javaList: java.util.List[T]): List[T] =
  javaList.asScala.toList

// Usage
val javaList: java.util.List[String] = ...
val scalaList: List[String] = javaListToScalaList(javaList)

But this is verbose and breaks the flow. What if you could write:

1
val scalaList = javaList.toScala  // Method appears on java.util.List!

This is extension methods via implicit classes. You’re not modifying java.util.List - you’re telling the compiler: “If someone calls .toScala on a Java List, secretly wrap it in my converter class.”

HOW implicit classes work

An implicit class is a class that the compiler automatically instantiates when it sees a method call that doesn’t exist on the original type.

The pattern:

1
2
3
4
5
implicit class MyExtensions[T](value: T) {
  def newMethod: SomeType = {
    // Use 'value' to compute result
  }
}

What the compiler does:

You write: someValue.newMethod
Compiler checks: Does someValue’s type have newMethod? No.
Compiler looks for implicit class that:
- Takes someValue’s type as constructor parameter
- Defines newMethod
Compiler rewrites your code as: new MyExtensions(someValue).newMethod

It’s syntactic sugar that makes code readable while keeping types separate.

sequenceDiagram
    actor Dev
    participant Compiler
    participant ImplicitClass as Implicit Class
    Dev->>Compiler: javaList.toScala
    Compiler->>Compiler: Check: does javaList have toScala?
    Compiler->>Compiler: No! Search for implicit class
    Compiler->>ImplicitClass: Found JavaToScalaList
    Compiler->>Compiler: Rewrite as: new JavaToScalaList(javaList).toScala
    ImplicitClass-->>Dev: Returns Scala List

WHAT the converters look like

1
2
3
4
implicit class JavaToScalaOption[T](javaOptional: Optional[T]) extends Serializable {
  def toOption: Option[T] =
    if (javaOptional.isPresent) Some(javaOptional.get()) else None
}

How this works:

implicit class tells the compiler this is an extension
[T] makes it generic - works for any type
(javaOptional: Optional[T]) constructor parameter - the value we’re extending
extends Serializable required for Spark (operations must be serializable)

Under the hood when you write:

1
2
val javaOpt: Optional[String] = Optional.of("hello")
val scalaOpt = javaOpt.toOption

The compiler transforms it to:

1
val scalaOpt = new JavaToScalaOption(javaOpt).toOption

You never see this transformation - but that’s what’s happening.

The converter suite

Java List → Scala List:

1
2
3
4
5
6
7
implicit class JavaToScalaList[E](javaArray: java.util.List[E]) extends Serializable {
  def toScala: List[E] = javaArray.asScala.toList
}

// Usage
val javaList: java.util.List[Int] = Arrays.asList(1, 2, 3)
val scalaList: List[Int] = javaList.toScala  // List(1, 2, 3)

Java Map → Scala Map:

1
2
3
4
5
6
7
8
implicit class JavaToScalaMap[K, V](javaMap: java.util.Map[K, V]) extends Serializable {
  def toScala: Map[K, V] = javaMap.asScala.toMap
}

// Usage
val javaMap: java.util.Map[String, Int] = new java.util.HashMap[String, Int]()
javaMap.put("count", 42)
val scalaMap: Map[String, Int] = javaMap.toScala  // Map("count" -> 42)

JSON parsing with implicits

This is where it gets interesting. We can extend Option[String] to parse JSON into case classes:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
implicit class VariousTypesFromJsonString(jsonString: Option[String]) extends Serializable {
  val mapper = new ObjectMapper() with ScalaObjectMapper
  mapper.registerModule(DefaultScalaModule)

  def jsonStringAs[T](failOnUnknownJsonProperties: Boolean = false)
                     (implicit m: Manifest[T]): Option[T] = {
    mapper.configure(DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES, failOnUnknownJsonProperties)
    jsonString.map(s => mapper.readValue[T](s))
  }

  def toMap[T](failOnUnknownJsonProperties: Boolean = false)
              (implicit m: Manifest[T]): Option[Map[String, T]] =
    jsonStringAs[Map[String, T]](failOnUnknownJsonProperties)
}

What’s happening:

Jackson ObjectMapper with Scala module handles JSON → Scala types
Manifest[T] captures the type at runtime (Scala’s reflection)
Option[String] makes it safe - if JSON is None, returns None
failOnUnknownJsonProperties controls strict vs lenient parsing

Usage:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
import com.vitthalmirji.dataengineering.common.Converters._

case class Employee(id: Int, name: String, age: Int)
case class Office(name: String, employees: Array[Employee])

val jsonString: Option[String] = Some("""
{
  "name": "Acme Corp",
  "employees": [
    {"id": 1, "name": "Alice", "age": 30},
    {"id": 2, "name": "Bob", "age": 25}
  ]
}
""")

// Parse JSON to case class
val office: Option[Office] = jsonString.jsonStringAs[Office]()

office match {
  case Some(o) => println(s"Office: ${o.name}, Employees: ${o.employees.length}")
  case None => println("Failed to parse JSON")
}

Why this matters in data pipelines:

Many data sources store metadata as JSON strings:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
// From a config table
val configJson: Option[String] = spark.table("configs")
  .filter($"key" === "pipeline_params")
  .select($"value")
  .as[String]
  .headOption

// Parse directly to case class
case class PipelineConfig(batchSize: Int, retries: Int, timeoutSec: Int)
val config: Option[PipelineConfig] = configJson.jsonStringAs[PipelineConfig]()

No manual parsing, no missing fields, type-safe.

Part 3 - String interpolators: Making date parsing readable

WHY dates are painful

Standard date parsing in Scala:

1
2
3
4
5
import java.time.{LocalDate, LocalDateTime}
import java.time.format.DateTimeFormatter

val date1 = LocalDate.parse("2025-06-28", DateTimeFormatter.ofPattern("yyyy-MM-dd"))
val date2 = LocalDateTime.parse("2025-06-28 14:30:00", DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss"))

After writing this 50 times, you realize:

You’re repeating the formatter every time
The format string is error-prone (MM vs mm? HH vs hh?)
It’s unreadable - the intent (parse a date) is buried

Quick Win

String interpolators aren’t just for dates - use them for any repeated pattern. SQL fragments with consistent escaping. S3/GCS paths with validation. Regex patterns with documented intent. If you’re copying the same string construction code 3+ times, write an interpolator. Five minutes now saves hours of debugging “why did that parse fail?” later.

What if it looked like:

1
2
val date1 = yearMonthDate"2025-06-28"
val date2 = yearMonthDate24HrTs"2025-06-28 14:30:00"

Clean. Readable. No formatter visible. That’s string interpolation.

HOW string interpolators work

You’ve seen s"Hello $name" - that’s Scala’s built-in interpolator. It rewrites:

1
2
3
s"Hello $name"
// into:
StringContext("Hello ", "").s(name)

Custom interpolators follow the same pattern:

Define an implicit class extending StringContext
Implement methods named after your interpolator
Use the StringContext to access the string parts

WHAT the date interpolator looks like

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
implicit class DateTimeInterpolator(sc: StringContext) {

  def yearMonthDate(args: Any*): LocalDate =
    LocalDate.parse(
      sc.s(args: _*).trim,
      DateTimeFormatter.ofPattern("yyyy-MM-dd")
    )

  def yearMonthDate24HrTs(args: Any*): LocalDateTime =
    LocalDateTime.parse(
      sc.s(args: _*).trim,
      DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss")
    )

  def yearMonthDate24HrNanoTs(args: Any*): LocalDateTime =
    LocalDateTime.parse(
      sc.s(args: _*).trim,
      DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss.SSS")
    )

  def tz(args: Any*): TimeZone =
    TimeZone.getTimeZone(sc.s(args: _*))
}

How it works step-by-step:

When you write:

1
val date = yearMonthDate"2025-06-28"

The compiler:

Sees a string with prefix yearMonthDate
Looks for an implicit StringContext class with method yearMonthDate
Finds DateTimeInterpolator
Rewrites as: new DateTimeInterpolator(StringContext("2025-06-28")).yearMonthDate()
Inside the method:
- sc.s(args: _*) reconstructs the string (handles interpolation if variables present)
- .trim removes whitespace
- LocalDate.parse(...) parses with the hardcoded format

With variables:

1
2
3
4
5
6
7
8
val year = 2025
val month = 6
val day = 28

val date = yearMonthDate"$year-$month-$day"
// Compiles to: new DateTimeInterpolator(StringContext("", "-", "-", "")).yearMonthDate(2025, 6, 28)
// sc.s(args) reconstructs: "2025-6-28"
// LocalDate.parse formats it properly

Usage in pipelines

Filtering by date:

1
2
3
4
5
6
7
8
9
import com.vitthalmirji.dataengineering.datetime.DateTimeInterpolators._

val startDate = yearMonthDate"2025-01-01"
val endDate = yearMonthDate"2025-06-28"

val filteredDf = transactionsDf.filter(
  $"transaction_date" >= lit(startDate.toString) &&
  $"transaction_date" <= lit(endDate.toString)
)

Timezone handling:

1
2
3
4
5
val centralTime = tz"US/Central"
val pacificTime = tz"US/Pacific"

// Convert timestamp with timezone
val zonedTs = yearMonthDate24HrNanoTsZone"2025-06-28 14:30:00.123 US/Pacific"

Comparing timestamps:

1
2
3
4
5
6
val cutoffTime = yearMonthDate24HrTs"2025-06-28 00:00:00"
val recordTime = yearMonthDate24HrTs"2025-06-28 14:30:00"

if (recordTime.isAfter(cutoffTime)) {
  println("Record is recent")
}

Why this matters

Readability: yearMonthDate"2025-06-28" vs LocalDate.parse(...)
Consistency: Everyone uses the same format, no mistakes
Compile-time safety: If the string doesn’t match the pattern, it fails at runtime immediately (not buried in logs)
Refactoring-friendly: Change the format in ONE place (the interpolator definition)

This pattern works for ANY domain-specific syntax: SQL fragments, URLs, regex, file paths. If you’re repeating string patterns, write an interpolator.

Part 3.5 - GCS String Interpolators: Working with Cloud Storage like a pro

WHY GCS operations are painful

Standard GCS operations in Scala require verbose Java SDK calls:

1
2
3
4
5
6
7
8
import com.google.cloud.storage.{Storage, StorageOptions, BlobId, Blob}

val storage = StorageOptions.getDefaultInstance.getService
val blobId = BlobId.of("my-bucket", "path/to/file.parquet")
val blob = storage.get(blobId)

// Get timestamp
val lastModified = blob.getUpdateTime

After writing this 100 times in data pipelines, you realize:

You’re repeating the storage service initialization
BlobId construction is verbose for every file access
No type safety - bucket/path typos fail at runtime
Listing blobs requires understanding Pages and iterators

What if it looked like:

1
2
3
val myBlob = blob"gs://my-bucket/path/to/file.parquet"
val timestamp = myBlob.updatedTs(None)
val allBlobs = blobs"gs://my-bucket/data/"

Clean. Readable. No SDK boilerplate. That’s GCS string interpolation.

HOW GCS interpolators work

Just like date interpolators, we extend StringContext to parse GCS URIs:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
implicit class GcsInterpolator(sc: StringContext) extends Serializable {
  def blob(args: Any*): Blob = {
    val gcsUri = sc.s(args: _*).trim
    val gcsObject = gcsUri.toGcsObject
    STORAGE_SERVICE.get(gcsObject.bucketName, gcsObject.objectName.get)
  }

  def blobs(args: Any*): Page[Blob] = {
    val gcsUri = sc.s(args: _*).trim
    val gcsObject = gcsUri.toGcsObject
    STORAGE_SERVICE.list(gcsObject.bucketName, BlobListOption.prefix(gcsObject.objectName.getOrElse("")))
  }

  def bucket(args: Any*): Bucket = {
    val bucketName = sc.s(args: _*).trim
    STORAGE_SERVICE.get(bucketName)
  }
}

What’s a GcsObject?

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
case class GcsObject(
  projectId: Option[String],
  bucketName: String,
  objectName: Option[String]
)

implicit class GcsObjectConversions(gcsURI: String) {
  def toGcsObject: GcsObject = {
    val tokens = if (gcsURI.startsWith("gs://")) {
      gcsURI.substring(5).split("/", 2)
    } else {
      gcsURI.split("/", 2)
    }

    if (tokens.isEmpty) throw new IllegalArgumentException(s"Invalid GCS URI $gcsURI")
    else if (tokens.tail.isEmpty) GcsObject(None, tokens.head, None)
    else GcsObject(None, tokens.head, Some(tokens.tail.mkString("/")))
  }
}

Parsing logic:

1
2
3
4
5
"gs://my-bucket/path/to/file.parquet".toGcsObject
// Result: GcsObject(None, "my-bucket", Some("path/to/file.parquet"))

"gs://my-bucket/".toGcsObject
// Result: GcsObject(None, "my-bucket", None)

WHAT the complete GCS interpolator suite looks like

blob - Single file access:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
def blob(args: Any*): Blob = {
  val totalString = sc.s(args: _*).trim
  val gcsObject = totalString.toGcsObject

  if (gcsObject.objectName.isEmpty) {
    throw new IllegalArgumentException("Path of object is empty")
  }

  val tryResult = trySafely(
    STORAGE_SERVICE.get(
      gcsObject.bucketName,
      gcsObject.objectName.get,
      BlobGetOption.fields(Storage.BlobField.values(): _*)
    ),
    Some(s"Error fetching blob $args")
  )
  tryResult.merge
}

blobs - List multiple files:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
def blobs(args: Any*)(implicit blobListOptions: Array[BlobListOption] = Array.empty): Page[Blob] = {
  val totalString = sc.s(args: _*).trim
  val gcsObject = totalString.toGcsObject
  val blobListOptionsWithDefaults = blobListOptions ++ Array(BlobListOption.pageSize(Int.MaxValue))

  val tryResult = trySafely(
    gcsObject.objectName.map(obj => {
      STORAGE_SERVICE.list(
        gcsObject.bucketName,
        Array(BlobListOption.prefix(obj)) ++ blobListOptionsWithDefaults: _*
      )
    }).getOrElse(STORAGE_SERVICE.list(gcsObject.bucketName, blobListOptionsWithDefaults: _*)),
    Some(s"Error fetching blobs $args")
  )
  tryResult.merge
}

manyBlobs - Multiple buckets/paths:

1
2
3
4
5
6
def manyBlobs(args: Any*): Map[GcsObject, Page[Blob]] = {
  val tokens = sc.s(args: _*).trim.split(",")
  tokens.map(gcsUri => {
    gcsUri.toGcsObject -> blobs"$gcsUri"
  }).toMap
}

bucket - Bucket access:

1
2
3
4
5
6
7
8
def bucket(args: Any*): Bucket = {
  val bucketName = sc.s(args: _*).trim
  val tryResult = trySafely(
    STORAGE_SERVICE.get(bucketName),
    Some(s"Error fetching bucket $bucketName")
  )
  tryResult.merge
}

buckets - List all buckets:

1
2
3
4
5
def buckets(args: Any*): List[Bucket] =
  trySafely(
    STORAGE_SERVICE.list().getValues.asScala.toList,
    Some("Error listing buckets")
  ).merge

flowchart TB
    A[GCS URI String] --> B{Parse with toGcsObject}
    B --> C[GcsObject]

    C --> D{Interpolator Type}

    D -->|blob| E["Single Blob<br/>blob&quot;gs://bucket/file&quot;"]
    D -->|blobs| F["Page of Blobs<br/>blobs&quot;gs://bucket/prefix&quot;"]
    D -->|manyBlobs| G["Map GcsObject → Page Blob<br/>manyBlobs&quot;gs://b1/,gs://b2/&quot;"]
    D -->|bucket| H["Single Bucket<br/>bucket&quot;my-bucket&quot;"]
    D -->|buckets| I["List of Buckets<br/>buckets&quot;&quot;"]

    E --> J[GCS Storage API]
    F --> J
    G --> J
    H --> J
    I --> J

    style J fill:#90EE90,stroke:#2d7a2d,color:#000

Usage in pipelines

Pattern 1: Check if file exists and get metadata:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
import com.vitthalmirji.dataengineering.gcp.GcsInterpolators._

val dataFile = blob"gs://my-bucket/data/2025-06-28/transactions.parquet"

if (dataFile != null && dataFile.exists()) {
  println(s"File size: ${dataFile.getSize} bytes")
  println(s"Last modified: ${dataFile.getUpdateTime}")
} else {
  println("File doesn't exist")
}

Pattern 2: List all files in a partition:

1
2
3
4
5
6
7
8
9
val partitionPath = "gs://my-bucket/warehouse/transactions/date=2025-06-28/"
val filesInPartition = blobs"$partitionPath"

val fileCount = filesInPartition.getValues.asScala.count(_.isFile)
println(s"Found $fileCount files in partition")

filesInPartition.getValues.asScala.foreach { blob =>
  println(s"File: ${blob.getName}, Size: ${blob.getSize}")
}

Pattern 3: Work with multiple buckets:

1
2
3
4
5
6
7
val sourceBuckets = "gs://source-bucket-1/,gs://source-bucket-2/,gs://source-bucket-3/"
val allBucketBlobs = manyBlobs"$sourceBuckets"

allBucketBlobs.foreach { case (gcsObject, blobPage) =>
  val count = blobPage.getValues.asScala.size
  println(s"Bucket ${gcsObject.bucketName}: $count objects")
}

Pattern 4: Get bucket info:

1
2
3
4
val myBucket = bucket"my-data-bucket"
println(s"Bucket location: ${myBucket.getLocation}")
println(s"Storage class: ${myBucket.getStorageClass}")
println(s"Created: ${myBucket.getCreateTime}")

Pattern 5: List all project buckets:

1
2
3
4
5
6
val allBuckets = buckets""
println(s"Total buckets in project: ${allBuckets.size}")

allBuckets.foreach { bucket =>
  println(s"- ${bucket.getName}")
}

Why this matters

Readability: blob"gs://bucket/file" vs 5 lines of SDK boilerplate
Type safety: Parse errors at string interpolation time
Consistency: Everyone uses the same GCS access pattern
Integration: Works seamlessly with affected partitions logic
Error handling: Built-in trySafely wrapping

Part 4 - Change Data Capture (CDC): SCD Type 1 with Spark

WHY CDC matters

In data warehousing, Change Data Capture tracks changes between dataset versions:

Inserts (new rows)
Updates (modified rows)
Deletes (missing rows)
No change (identical rows)

Slowly Changing Dimension Type 1 (SCD Type 1) means: overwrite old values with new ones, don’t keep history.

Use case: You have a customers table. Every night:

Load fresh customer data from source
Compare with yesterday’s table
Mark each row as: I (insert), U (update), D (delete), or NC (no change)
Write the delta

HOW Delta CDC works

The algorithm:

Compute a hash of all non-key columns for both datasets
Full outer join on primary keys
Compare hashes to detect changes:
- If left-only: Insert
- If right-only: Delete
- If hashes match: No change
- If hashes differ: Update

Why hash instead of column-by-column comparison?

Performance: One hash comparison vs N column comparisons
Simplicity: Hash mismatch = something changed, don’t care what
Deterministic: Same data → same hash

flowchart TB
    A[Source DataFrame] --> B[Compute Hash]
    C[Previous DataFrame] --> D[Compute Hash]

    B --> E{Full Outer Join<br/>on Primary Keys}
    D --> E

    E -->|Left Only| F[Insert - I]
    E -->|Right Only| G[Delete - D]
    E -->|Both: Hash Match| H[No Change - NC]
    E -->|Both: Hash Differs| I[Update - U]

    F --> J[Delta DataFrame]
    G --> J
    H --> J
    I --> J

    style F fill:#90EE90,stroke:#2d7a2d,color:#000
    style G fill:#FFB6C6,stroke:#cc0066,color:#000
    style H fill:#87CEEB,stroke:#0066cc,color:#000
    style I fill:#FFD700,stroke:#cc8800,color:#000

WHAT the implementation looks like

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
implicit class DeltaTransformations(dataframe: DataFrame) extends Serializable {

  def performDelta(
    previousLoadedDataset: Option[DataFrame],
    primaryKeys: List[String],
    columnsExcludeHashcodeCalculation: Option[List[String]] = None
  ): DataFrame = {

    val deltaDf = previousLoadedDataset.map { previousDataset =>
      if (primaryKeys.isEmpty) {
        throw new IllegalArgumentException("Primary keys cannot be empty")
      }

      // Step 1: Determine which columns to hash
      val newDatasetColumns = dataframe.columns
      val hashColumnsList = newDatasetColumns.filterNot(c =>
        (primaryKeys ++ AUDIT_COLUMN_NAMES ++ columnsExcludeHashcodeCalculation.getOrElse(List())).contains(c)
      )

      // Step 2: Build join condition on primary keys
      var joinExpr: Column = lit(true)
      primaryKeys.foreach(k => joinExpr = (col(s"src.$k") === col(s"stg.$k")) && joinExpr)

      // Step 3: Add hashcode column to both datasets
      val sourceDf = dataframe.withColumn("hashcode", hash64(concat_ws("~", hashColumnsList.map(col): _*)))
      val stageDf = previousDataset
        .filter("`delta_flag` <> 'D'")  // Exclude previously deleted rows
        .drop("delta_flag")
        .withColumn("hashcode", hash64(concat_ws("~", hashColumnsList.map(col): _*)))

      // Step 4: Full outer join
      val deltaDf = sourceDf.as("src")
        .join(stageDf.as("stg"), joinExpr, "full_outer")
        .withColumn("delta_flag",
          when(col("stg.id").isNull, "I")              // Left-only = Insert
          .when(col("src.id").isNull, "D")             // Right-only = Delete
          .when(col("src.hashcode") === col("stg.hashcode"), "NC")  // Hash match = No change
          .otherwise("U")                              // Hash differ = Update
        )
        .select((previousLoadedDatasetColumns ++ Array("delta_flag")).map(resolveAuditColumns): _*)

      deltaDf.selectExpr(newDatasetColumns: _*)
    }

    // If no previous dataset (initial load), mark everything as Insert
    deltaDf.getOrElse(dataframe.withColumn("delta_flag", lit("I")))
  }

  // Hash function using Spark's XxHash64
  private def hash64(cols: Column*): Column =
    new Column(new XxHash64(cols.map(_.expr)))
}

Step-by-step breakdown:

1. Filter hash columns:

Exclude primary keys (they’re in the join condition)
Exclude audit columns (load_ts, upd_ts) - they change every run
Exclude user-specified columns

2. Build join expression:

1
2
// For primaryKeys = List("id", "date")
// Becomes: (src.id === stg.id) && (src.date === stg.date) && true

3. Compute hashes:

1
2
3
// Concat all hash columns with "~" separator
// Example: "Alice~30~USA" → hash64 → 8473298473298472
hash64(concat_ws("~", col("name"), col("age"), col("country")))

Why concat_ws("~", ...)?

Concatenates columns with separator
Hash of concatenated string is deterministic
Separator prevents collisions: "AB", "C" vs "A", "BC" → different hashes

4. Full outer join:

src.id	src.name	src.hash	stg.id	stg.name	stg.hash	delta_flag
1	Alice	abc123	1	Alice	abc123	NC
2	Bob	def456	2	Robert	xyz789	U
3	Carol	ghi789	NULL	NULL	NULL	I
NULL	NULL	NULL	4	Dave	jkl012	D

5. Delta logic:

Insert: Source exists, stage doesn’t → New row
Delete: Stage exists, source doesn’t → Row removed
No Change: Hashes match → Data identical
Update: Hashes differ → Data changed

Usage example

Initial load:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
import com.vitthalmirji.dataengineering.spark.ETL._

val sourceDf = spark.table("source.customers")
  .withColumn("load_ts", current_timestamp)
  .withColumn("upd_ts", current_timestamp)

val deltaDf = sourceDf.performDelta(
  previousLoadedDataset = None,  // No previous data
  primaryKeys = List("customer_id")
)

deltaDf.write.mode("overwrite").saveAsTable("warehouse.customers")

// All rows have delta_flag = 'I'

Incremental load:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
val sourceDf = spark.table("source.customers")
val previousDf = spark.table("warehouse.customers")

val deltaDf = sourceDf.performDelta(
  previousLoadedDataset = Some(previousDf),
  primaryKeys = List("customer_id"),
  columnsExcludeHashcodeCalculation = Some(List("last_login_ts"))  // Exclude volatile columns
)

// Overwrite only affected partitions
deltaDf.write.mode("overwrite").insertInto("warehouse.customers")

// Validate
spark.table("warehouse.customers")
  .groupBy("delta_flag")
  .count()
  .show()
// +----------+-----+
// |delta_flag|count|
// +----------+-----+
// |I         |120  |
// |U         |45   |
// |D         |10   |
// |NC        |5000 |
// +----------+-----+

Part 5 - Delta Lake utilities: Vacuum, compaction, and Z-ordering

WHY Delta Lake needs maintenance

Delta Lake stores multiple versions of your data for time travel. Over time:

Old files accumulate (every write creates new Parquet files)
Small files proliferate (streaming inserts create many tiny files)
Query performance degrades (reading 1000 small files vs 10 large files)

You need to:

Vacuum: Delete old files no longer needed for time travel
Compact: Merge small files into larger ones
Z-Order: Co-locate related data for faster reads

Doing this manually:

1
2
3
4
5
6
7
8
// Vacuum (delete files older than 7 days)
spark.sql("VACUUM warehouse.customers RETAIN 168 HOURS")

// Compaction
spark.sql("OPTIMIZE warehouse.customers")

// Z-ordering
spark.sql("OPTIMIZE warehouse.customers ZORDER BY (customer_id, region)")

Problems:

Verbose, repetitive
No type safety (typo in table name = runtime error)
No error handling
Hard to integrate into pipeline code

HOW to wrap Delta ops

We create utility functions that:

Take table name as parameter
Handle errors gracefully
Return Boolean for success/failure
Enable optional partition filtering

WHAT the Delta utilities look like

Vacuum:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
object DeltaTableUtils {

  def executeVacuum(
    tableName: String,
    retentionHours: Double = 168.0  // 7 days default
  ): Boolean = {

    val spark: SparkSession = SparkSession.getActiveSession.get
    spark.conf.set("spark.databricks.delta.vacuum.parallelDelete.enabled", "true")

    try {
      val result = DeltaTable
        .forName(tableName)
        .vacuum(retentionHours)

      if (result.isEmpty) {
        logger.info(s"Vacuum successful for $tableName, retention=$retentionHours hours")
        true
      } else {
        logger.error(s"Vacuum failed for $tableName")
        false
      }
    } catch {
      case e: Throwable =>
        logger.error(s"Vacuum over $tableName failed: ${e.getMessage}")
        throw e
    }
  }
}

What’s happening:

Enable parallel delete for faster vacuuming
Use Delta Lake’s DeltaTable.forName() API (type-safe table lookup)
Call .vacuum(retentionHours) - deletes files older than threshold
Return success/failure

WHY vacuum matters:

Delta keeps files for time travel (SELECT * FROM table VERSION AS OF 5)
Old files consume storage
Vacuum deletes files no longer needed
Warning: After vacuum, you can’t query versions older than retention period

Compaction:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
def executeCompaction(
  tableName: String,
  partitionFilterExpr: Option[String] = None
): Boolean = {

  try {
    val optimizeDf = DeltaTable
      .forName(tableName)
      .optimize()

    partitionFilterExpr
      .map(filterExpr => optimizeDf.where(filterExpr))
      .getOrElse(optimizeDf)
      .executeCompaction()

    true
  } catch {
    case t: Throwable =>
      logger.error(s"Compaction failed for $tableName: ${t.getMessage}")
      throw t
  }
}

What’s happening:

Call .optimize() on Delta table
Optionally filter to specific partitions (e.g., "date >= '2025-06-01'")
Execute compaction - merges small files into larger ones

WHY compaction matters:

Streaming writes create many small files
Reading 1000 small files is slow (file open overhead)
Compaction merges them into ~1GB files
Result: Queries run 5-10x faster

Z-Ordering:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
def executeZOrderBy(
  tableName: String,
  zOrderByKeys: Option[Seq[String]] = None,
  partitionFilterExpr: Option[String] = None
): Boolean = {

  try {
    if (zOrderByKeys.isDefined && zOrderByKeys.get.isEmpty) {
      logger.error(s"Invalid zOrderBy keys: ${zOrderByKeys.get}")
      return false
    }

    val optimizeDf = DeltaTable
      .forName(tableName)
      .optimize()

    val toCompactDf = partitionFilterExpr
      .map(filterExpr => optimizeDf.where(filterExpr))
      .getOrElse(optimizeDf)

    zOrderByKeys
      .map(keys => toCompactDf.executeZOrderBy(keys: _*))
      .getOrElse(toCompactDf.executeZOrderBy())

    true
  } catch {
    case t: Throwable =>
      logger.error(s"Z-ordering failed for $tableName: ${t.getMessage}")
      throw t
  }
}

What’s happening:

Validate Z-order keys are provided
Call .optimize() on Delta table
Optionally filter to specific partitions
Execute Z-ordering on specified columns

Performance win

Z-ordering isn’t magic, but the wins are real. One team Z-ordered their 5TB transactions table by customer_id and date - queries that scanned 500GB before now scan 50GB. Same query, 10x less data read, 70% faster execution. The catch: only works if your queries actually filter on those Z-order columns. Profile first, then optimize.

WHY Z-ordering matters:

Z-ordering is a multi-dimensional clustering technique. It co-locates related data in the same files.

Example: You have a transactions table with columns customer_id, date, amount.

Without Z-ordering:

File 1: customers A, B, C from dates 2025-01-01 to 2025-06-30
File 2: customers D, E, F from dates 2025-01-01 to 2025-06-30
Query: WHERE customer_id = 'A' AND date = '2025-06-15'
Reads: Both files (because customer A is in file 1, but scattered across many row groups)

With Z-ordering on (customer_id, date):

File 1: customer A, dates 2025-01-01 to 2025-03-31
File 2: customer A, dates 2025-04-01 to 2025-06-30
File 3: customer B, dates 2025-01-01 to 2025-03-31
Query: WHERE customer_id = 'A' AND date = '2025-06-15'
Reads: Only file 2 (customer A’s June data is co-located)

Result: 70-90% less data scanned on filtered queries.

graph LR
    A[Without Z-Ordering] --> B[File 1: Customers A-F<br/>Dates: Jan-Jun]
    A --> C[File 2: Customers G-M<br/>Dates: Jan-Jun]

    D[With Z-Ordering<br/>on customer_id, date] --> E[File 1: Customer A<br/>Dates: Jan-Mar]
    D --> F[File 2: Customer A<br/>Dates: Apr-Jun]
    D --> G[File 3: Customer B<br/>Dates: Jan-Mar]

    H[Query: WHERE customer_id='A'<br/>AND date='2025-06-15'] --> B
    H --> C
    I[Same Query with Z-Order] --> F

    style B fill:#FFB6C6,stroke:#cc0066,color:#000
    style C fill:#FFB6C6,stroke:#cc0066,color:#000
    style F fill:#90EE90,stroke:#2d7a2d,color:#000

    B -.->|Reads both files| J[Slow: 2 files]
    C -.->|Reads both files| J
    F -.->|Reads 1 file| K[Fast: 1 file]

    style J fill:#FFB6C6,stroke:#cc0066,color:#000
    style K fill:#90EE90,stroke:#2d7a2d,color:#000

Usage in pipelines

Pattern 1: Vacuum after batch load

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
import com.vitthalmirji.dataengineering.spark.DeltaTableUtils._

// Load data
sourceDf.write.mode("append").saveAsTable("warehouse.transactions")

// Vacuum old versions (keep last 7 days)
val vacuumSuccess = executeVacuum("warehouse.transactions", retentionHours = 168.0)

if (!vacuumSuccess) {
  logger.warn("Vacuum failed - old files still present, but load succeeded")
}

Pattern 2: Compact on schedule

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
// Run compaction nightly for today's partition
val today = LocalDate.now().toString

val compactSuccess = executeCompaction(
  tableName = "warehouse.transactions",
  partitionFilterExpr = Some(s"date = '$today'")
)

if (compactSuccess) {
  logger.info("Compaction complete - small files merged")
}

Pattern 3: Z-order on commonly filtered columns

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
// Z-order by customer_id and date (common filter combo)
val zOrderSuccess = executeZOrderBy(
  tableName = "warehouse.transactions",
  zOrderByKeys = Some(Seq("customer_id", "date")),
  partitionFilterExpr = Some(s"date >= '2025-01-01'")  // Only recent data
)

if (zOrderSuccess) {
  logger.info("Z-ordering complete - queries on customer_id/date will be faster")
}

Pattern 4: Combined maintenance workflow

flowchart TB
    A[Delta Table with Issues] --> B[1000s of small files<br/>Slow queries<br/>Old versions]

    B --> C[Step 1: Compaction]
    C --> D[Merge small files<br/>→ ~1GB files]

    D --> E[Step 2: Z-Ordering]
    E --> F[Co-locate related data<br/>by customer_id, date]

    F --> G[Step 3: Vacuum]
    G --> H[Delete files older<br/>than 7 days]

    H --> I[Optimized Table]
    I --> J[✓ Fewer files<br/>✓ Faster queries<br/>✓ Less storage]

    style A fill:#FFB6C6,stroke:#cc0066,color:#000
    style I fill:#90EE90,stroke:#2d7a2d,color:#000
    style J fill:#90EE90,stroke:#2d7a2d,color:#000

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
def maintainDeltaTable(tableName: String, zOrderCols: Seq[String]): Unit = {
  logger.info(s"Starting maintenance for $tableName")

  // Step 1: Compact small files
  if (!executeCompaction(tableName)) {
    throw new RuntimeException(s"Compaction failed for $tableName")
  }

  // Step 2: Z-order for query performance
  if (!executeZOrderBy(tableName, Some(zOrderCols))) {
    throw new RuntimeException(s"Z-ordering failed for $tableName")
  }

  // Step 3: Vacuum old files
  if (!executeVacuum(tableName, retentionHours = 168.0)) {
    logger.warn(s"Vacuum failed for $tableName - old files remain")
  }

  logger.info(s"Maintenance complete for $tableName")
}

// Usage
maintainDeltaTable("warehouse.customers", Seq("customer_id", "region"))
maintainDeltaTable("warehouse.transactions", Seq("customer_id", "date"))

Part 6 - Schema evolution: Automatic column add/drop with validation

WHY schema evolution is hard

Production tables change over time:

New columns added (source adds phone_number field)
Columns removed (deprecated fax_number field)
Primary keys must stay stable (can’t change identity)
Partitions must stay stable (can’t change partitioning scheme)

Manual schema evolution:

1
2
3
4
5
6
7
8
// Add column manually
spark.sql("ALTER TABLE warehouse.customers ADD COLUMN phone_number STRING")

// Backfill existing rows with NULL
spark.sql("UPDATE warehouse.customers SET phone_number = NULL WHERE phone_number IS NULL")

// Drop column manually
spark.sql("ALTER TABLE warehouse.customers DROP COLUMN fax_number")

Problems:

No validation - you can accidentally drop a primary key
No automation - you manually detect schema changes
No history - you can’t roll back if something breaks
Error-prone - typos, wrong table, wrong column

HOW to automate schema evolution

We need utilities that:

Detect schema changes - compare source vs target schemas
Validate changes - ensure primary keys don’t change
Evolve automatically - add new columns, drop old ones
Handle partitions - use different strategies for partitioned tables

WHAT the schema comparison utilities look like

Detect schema changes:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
object SchemaOps {

  def isSchemaChanged(df1: DataFrame, df2: DataFrame): Boolean = {
    val map1 = getSchemaMap(df1)
    val map2 = getSchemaMap(df2)
    !map1.equals(map2)
  }

  def getSchemaMap(df: DataFrame): Map[String, String] = {
    var map = Map.empty[String, String]
    df.schema.foreach(f => map += f.name -> f.dataType.simpleString)
    map
  }
}

What’s happening:

Convert DataFrame schema to Map[columnName, dataType]
Compare maps - if they differ, schema changed

Usage:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
import com.vitthalmirji.dataengineering.schema.SchemaOps._

val sourceDf = spark.table("source.customers")
val targetDf = spark.table("warehouse.customers")

if (isSchemaChanged(sourceDf, targetDf)) {
  logger.warn("Schema change detected - initiating evolution")
  // Trigger schema evolution...
} else {
  logger.info("No schema changes - proceeding with normal load")
}

Detect new columns:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
def newColumnsAdded(oldDf: DataFrame, newDf: DataFrame): Boolean =
  !newDf.columns.filterNot(oldDf.columns.toSet).isEmpty

def getNewColumnsDf(
  oldDf: DataFrame,
  newDf: DataFrame,
  primaryKeys: Seq[String] = Seq()
): DataFrame = {
  val newDfCols = newDf.columns.filterNot(primaryKeys.toSet)
  val oldDfCols = oldDf.columns.filterNot(primaryKeys.toSet)
  val cols = newDfCols.filterNot(oldDfCols.toSet).toList ++ primaryKeys
  newDf.select(cols.map(c => col(c)): _*)
}

What’s happening:

Find columns in newDf but not in oldDf
Return DataFrame with only new columns + primary keys

WHY this matters:

When adding columns, you need to backfill existing rows. Instead of updating in place:

1
2
// ❌ BAD: Full table update (slow, locks table)
spark.sql("UPDATE warehouse.customers SET phone_number = NULL")

We join new columns with existing data:

1
2
3
4
5
6
7
8
9
// ✅ GOOD: Join approach (fast, no locks)
val newColsDf = getNewColumnsDf(targetDf, sourceDf, primaryKeys = Seq("customer_id"))

val evolvedDf = targetDf
  .join(newColsDf, Seq("customer_id"), "left")
  .selectExpr(sourceDf.columns: _*)  // Reorder to match source schema

// Write back
evolvedDf.write.mode("overwrite").saveAsTable("warehouse.customers")

WHAT the schema evolution workflow looks like

For non-partitioned tables:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
case class SchemaEvolution(
  sourceDf: DataFrame,
  targetTable: String,
  primaryKeys: Seq[String]
)

def evolveNonPartitionedTable(input: SchemaEvolution): Unit = {
  val targetDf = spark.table(input.targetTable)

  // Step 1: Detect changes
  val changeMap = getChangeColumns(targetDf, input.sourceDf)
  val columnsToAdd = changeMap.getOrElse("COLS_TO_ADD", Seq())
  val columnsToDrop = changeMap.getOrElse("COLS_TO_DROP", Seq())

  if (columnsToAdd.isEmpty && columnsToDrop.isEmpty) {
    throw new Exception("No schema changes detected")
  }

  // Step 2: Validate primary keys didn't change
  if (input.primaryKeys.exists(columnsToAdd.contains) ||
      input.primaryKeys.exists(columnsToDrop.contains)) {
    throw new Exception("Primary key changes not allowed")
  }

  // Step 3: Build evolved DataFrame
  val evolvedDf = if (columnsToAdd.nonEmpty) {
    val newColsDf = getNewColumnsDf(targetDf, input.sourceDf, input.primaryKeys)
    targetDf.join(newColsDf, input.primaryKeys, "left")
  } else {
    targetDf  // Just dropping columns, no join needed
  }

  // Step 4: Write with overwriteSchema
  evolvedDf
    .selectExpr(input.sourceDf.columns: _*)
    .write
    .mode("overwrite")
    .option("overwriteSchema", "true")
    .saveAsTable(input.targetTable)

  logger.info(s"Schema evolution complete for ${input.targetTable}")
}

def getChangeColumns(
  tgtDf: DataFrame,
  sourceDf: DataFrame
): Map[String, Seq[String]] = {
  val sourceCols = sourceDf.columns
  val targetCols = tgtDf.columns
  val newCols = sourceCols.filterNot(targetCols.toSet).toList
  val droppedCols = targetCols.filterNot(sourceCols.toSet).toList
  Map("COLS_TO_ADD" -> newCols, "COLS_TO_DROP" -> droppedCols)
}

What’s happening:

Detect changes: Compare source vs target schemas
Validate: Ensure primary keys aren’t touched
Add columns: Join new columns from source (NULLs backfilled)
Drop columns: Select only source columns (effectively drops extra columns)
Write with overwriteSchema: Tells Delta to accept schema changes

Schema evolution danger

Never evolve partition columns or primary keys in production. If you add/drop/rename a partition column, every downstream job reading that table breaks. If you change primary keys, CDC logic produces garbage. Schema evolution is for adding business columns, not changing table structure. Changing structure means creating a new table and migrating.

For partitioned tables:

Partitioned tables need special handling:

Can’t use overwriteSchema on partitioned writes (Delta limitation)
Must use mergeSchema when adding columns
Must use ALTER TABLE when dropping columns

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
def evolvePartitionedTable(
  input: SchemaEvolution,
  partitionCols: Seq[String]
): Unit = {

  val targetDf = spark.table(input.targetTable)
  val changeMap = getChangeColumns(targetDf, input.sourceDf)
  val columnsToAdd = changeMap.getOrElse("COLS_TO_ADD", Seq())
  val columnsToDrop = changeMap.getOrElse("COLS_TO_DROP", Seq())

  // Validate partitions didn't change
  if (partitionCols.exists(columnsToAdd.contains) ||
      partitionCols.exists(columnsToDrop.contains)) {
    throw new Exception("Partition column changes not allowed")
  }

  // Validate primary keys include partitions
  if (!partitionCols.forall(input.primaryKeys.contains)) {
    throw new Exception("Partition columns must be part of primary keys")
  }

  // Add columns: Use mergeSchema
  if (columnsToAdd.nonEmpty) {
    val newColsDf = getNewColumnsDf(targetDf, input.sourceDf, input.primaryKeys)
    val evolvedDf = targetDf.join(newColsDf, input.primaryKeys, "left")

    evolvedDf
      .selectExpr(input.sourceDf.columns: _*)
      .write
      .mode("overwrite")
      .partitionBy(partitionCols: _*)
      .option("mergeSchema", "true")
      .saveAsTable(input.targetTable)
  }

  // Drop columns: Use ALTER TABLE
  if (columnsToDrop.nonEmpty) {
    // Enable column mapping (required for column drops)
    spark.sql(s"""
      ALTER TABLE ${input.targetTable}
      SET TBLPROPERTIES (
        'delta.minReaderVersion' = '2',
        'delta.minWriterVersion' = '5',
        'delta.columnMapping.mode' = 'name'
      )
    """)

    // Drop columns
    val dropList = columnsToDrop.mkString(",")
    spark.sql(s"ALTER TABLE ${input.targetTable} DROP COLUMNS ($dropList)")
  }

  logger.info(s"Partitioned schema evolution complete for ${input.targetTable}")
}

Why different strategies?

Adding columns to partitioned table: Can’t use overwriteSchema (breaks partition evolution), must use mergeSchema
Dropping columns: Delta needs column mapping enabled first (so it can track columns by ID, not name)
Primary keys must include partitions: Otherwise you can’t uniquely identify rows within a partition

flowchart TB
    A[Source DataFrame] --> B{isSchemaChanged?}
    B -->|No| C[Normal Load]
    B -->|Yes| D[getChangeColumns]

    D --> E{Validate}
    E -->|Primary Keys Changed| F[ABORT:<br/>Not Allowed]
    E -->|Partitions Changed| F
    E -->|Valid Changes| G{Column Changes}

    G -->|Add Columns| H[Join new columns<br/>from source]
    G -->|Drop Columns| I[Select only<br/>source columns]
    G -->|Both| J[Join + Select]

    H --> K{Partitioned?}
    I --> K
    J --> K

    K -->|No| L[Write with<br/>overwriteSchema]
    K -->|Yes - Add| M[Write with<br/>mergeSchema]
    K -->|Yes - Drop| N[ALTER TABLE<br/>DROP COLUMNS]

    L --> O[Schema Evolved]
    M --> O
    N --> O

    style F fill:#FFB6C6,stroke:#cc0066,color:#000
    style O fill:#90EE90,stroke:#2d7a2d,color:#000

Usage in pipelines

Pattern 1: Auto-detect and evolve

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
import com.vitthalmirji.dataengineering.schema.SchemaOps._

val sourceDf = spark.table("source.customers")
val targetTable = "warehouse.customers"
val targetDf = spark.table(targetTable)

if (isSchemaChanged(sourceDf, targetDf)) {
  logger.warn("Schema change detected")

  val input = SchemaEvolution(
    sourceDf = sourceDf,
    targetTable = targetTable,
    primaryKeys = Seq("customer_id")
  )

  evolveNonPartitionedTable(input)
} else {
  // Normal load
  sourceDf.write.mode("overwrite").saveAsTable(targetTable)
}

Pattern 2: Validate before evolution

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
val sourceDf = spark.table("source.transactions")
val targetTable = "warehouse.transactions"
val targetDf = spark.table(targetTable)

if (isSchemaChanged(sourceDf, targetDf)) {
  val changeMap = getChangeColumns(targetDf, sourceDf)
  val columnsToAdd = changeMap("COLS_TO_ADD")
  val columnsToDrop = changeMap("COLS_TO_DROP")

  logger.info(s"Columns to add: ${columnsToAdd.mkString(", ")}")
  logger.info(s"Columns to drop: ${columnsToDrop.mkString(", ")}")

  // Alert on significant changes
  if (columnsToDrop.nonEmpty) {
    logger.warn("Column drops detected - review before proceeding")
    // Send alert, require manual approval, etc.
  }

  // Evolve
  val input = SchemaEvolution(
    sourceDf = sourceDf,
    targetTable = targetTable,
    primaryKeys = Seq("transaction_id", "date")
  )

  evolvePartitionedTable(input, partitionCols = Seq("date"))
}

Part 7 - Data quality wrappers: Integrating DQ frameworks

WHY data quality checks matter

Production pipelines need guardrails:

Schema validation (expected columns present?)
Null checks (critical fields non-null?)
Value ranges (age between 0-120?)
Referential integrity (foreign keys exist?)
Duplicate detection (primary keys unique?)

Manual checks:

1
2
3
4
5
6
7
8
9
val nullCount = df.filter($"customer_id".isNull).count()
if (nullCount > 0) {
  throw new RuntimeException(s"Found $nullCount null customer_ids")
}

val dupCount = df.groupBy("customer_id").count().filter($"count" > 1).count()
if (dupCount > 0) {
  throw new RuntimeException(s"Found $dupCount duplicate customer_ids")
}

Problems:

Verbose, repetitive
Hard to maintain (adding checks = adding code)
No centralized config (rules scattered across pipelines)
No reporting (can’t see DQ trends over time)

DQ integration note

The DQ wrapper pattern works with any rule engine - Great Expectations, Deequ, custom frameworks, or commercial tools. The key is centralizing rule configuration (database, config files, whatever) and making execution one-line (.executeDQ). Swap out the rule engine implementation without changing pipeline code. Pick what fits your stack.

Solution: Data Quality frameworks (like Great Expectations, Deequ, or proprietary DQ libraries) let you define rules centrally and execute them declaratively.

But integrating these frameworks still requires boilerplate:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
val ruleEngine = new RuleEngine()
ruleEngine.registerRuleset(ruleSetId = 200)
val results = ruleEngine.executeRules(df)
val isSuccess = !results.exists(_.getFailedCount > 0)

if (!isSuccess) {
  // Send email
  // Log failures
  // Throw exception
}

HOW to wrap DQ frameworks

We use implicit classes to add .executeDQ() method to DataFrames:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
implicit class DataQualityActions(dataframe: Dataset[Row]) extends Serializable {

  def executeDQ(
    ruleSetId: Int,
    emailDetails: Option[Email] = None,
    emailOnSuccess: Boolean = true,
    tableName: Option[String] = None
  ): Boolean = {
    // Execute rules and return success/failure
  }
}

Now any DataFrame can call:

1
2
3
4
val isValid = df.executeDQ(ruleSetId = 200)
if (!isValid) {
  throw new RuntimeException("DQ checks failed")
}

WHAT the DQ wrapper looks like

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
object DataQuality {

  val ruleEngine = new RuleEngine()  // Your DQ framework

  implicit class DataQualityActions(dataframe: Dataset[Row]) extends Serializable {

    def executeDQ(
      ruleSetId: Int,
      emailDetails: Option[Email] = None,
      emailOnSuccess: Boolean = true,
      tableName: Option[String] = None
    ): Boolean = {
      // Register rules from centralized config
      ruleEngine.registerRuleset(ruleSetId, Environment.PROD)

      // Execute all rules against DataFrame
      val metrics = ruleEngine.executeRules(dataframe)

      // Check if any rule failed
      val isSuccess = !metrics.exists(_.getFailedCount > 0)

      // Send email if configured
      if (emailDetails.isDefined && (!isSuccess || (isSuccess && emailOnSuccess))) {
        val rules = ruleEngine.getRules
        ruleEngine.generateAndSendEmail(emailDetails.get, rules, metrics, tableName)
      }

      // Clear rules for next run
      ruleEngine.clearRules()

      isSuccess
    }
  }
}

What’s happening:

Register ruleset: Load rules from centralized config (e.g., rule ID 200 = “customer validation rules”)
Execute rules: Run all rules against the DataFrame
Check results: Any rule with failed_count > 0 means DQ failed
Send email: Optionally email results (always on failure, optionally on success)
Clear rules: Clean up for next execution
Return Boolean: True = all passed, False = at least one failed

WHY this design?

Implicit class: Makes DQ checks read like native Spark operations
Rule ID: Centralized config (change rules without code changes)
Email integration: Built-in reporting
Boolean return: Simple pass/fail for pipeline control flow

sequenceDiagram
    participant Pipeline
    participant DataFrame
    participant DQ as DataQuality.executeDQ
    participant Engine as RuleEngine
    participant Email as EmailService
    participant DB as AuditDB

    Pipeline->>DataFrame: transformedDf
    DataFrame->>DQ: .executeDQ(ruleSetId=200)
    DQ->>Engine: registerRuleset(200)
    Engine-->>DQ: Rules loaded
    DQ->>Engine: executeRules(dataframe)
    Engine-->>DQ: metrics

    alt All Rules Pass
        DQ->>Email: Send success email (optional)
        DQ->>DB: Write metrics (optional)
        DQ-->>Pipeline: true
        Pipeline->>Pipeline: Write to warehouse
    else Any Rule Fails
        DQ->>Email: Send failure email
        DQ->>DB: Write metrics (optional)
        DQ-->>Pipeline: false
        Pipeline->>Pipeline: Write to staging/abort
    end

Usage in pipelines

Pattern 1: Basic DQ check

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
import com.vitthalmirji.dataengineering.spark.DataQuality._

val transformedDf = sourceDf
  .filter($"amount" > 0)
  .groupBy($"customer_id")
  .agg(sum($"amount").alias("total"))

// Execute DQ checks (ruleset 200 = customer totals validation)
val isValid = transformedDf.executeDQ(ruleSetId = 200)

if (isValid) {
  transformedDf.write.mode("overwrite").saveAsTable("warehouse.customer_totals")
} else {
  throw new RuntimeException("DQ checks failed - NOT writing to warehouse")
}

Pattern 2: DQ with email alerts

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
import com.vitthalmirji.dataengineering.spark.DataQuality._
import com.vitthalmirji.dataengineering.common.Email

val email = Email(
  from = "[email protected]",
  to = List("[email protected]"),
  subject = Some("DQ Report: Customer Totals"),
  body = None  // Auto-generated by DQ framework
)

val isValid = transformedDf.executeDQ(
  ruleSetId = 200,
  emailDetails = Some(email),
  emailOnSuccess = false,  // Only email on failure
  tableName = Some("warehouse.customer_totals")
)

if (!isValid) {
  logger.error("DQ failed - check email for details")
  throw new RuntimeException("DQ checks failed")
}

Pattern 3: DQ with fallback (write to staging on failure)

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
val isValid = transformedDf.executeDQ(ruleSetId = 200)

if (isValid) {
  // Write to production
  transformedDf.write.mode("overwrite").saveAsTable("warehouse.customers")
  logger.info("DQ passed - written to warehouse")
} else {
  // Write to staging for investigation
  transformedDf.write.mode("overwrite").saveAsTable("staging.customers_failed_dq")
  logger.warn("DQ failed - written to staging for review")
}

Pattern 4: DQ with database persistence

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
def executeDQWithDBWrite(
  ruleSetId: Int,
  emailDetails: Option[Email] = None,
  emailOnSuccess: Boolean = false,
  targetTable: String,
  dbProps: Properties,
  batchId: Long = 0
): Boolean = {

  val (isSuccess, rules, metrics) = getDQResult(
    ruleSetId = ruleSetId,
    emailDetails = emailDetails,
    emailOnSuccess = emailOnSuccess,
    tableName = Some(targetTable)
  )

  // Write DQ results to audit database
  writeDQResultToDB(targetTable, metrics, rules, dbProps, batchId)

  isSuccess
}

// Usage
val dbProps = new Properties()
dbProps.setProperty("url", "jdbc:postgresql://localhost:5432/audit")
dbProps.setProperty("user", "audit_writer")
dbProps.setProperty("password", "***")

val isValid = transformedDf.executeDQWithDBWrite(
  ruleSetId = 200,
  targetTable = "warehouse.customers",
  dbProps = dbProps,
  batchId = 12345
)

// Later: Query audit DB to see DQ trends over time
spark.read.jdbc(dbProps, "dq_audit_table")
  .filter($"target_table" === "warehouse.customers")
  .groupBy("run_date")
  .agg(sum("failed_count").alias("total_failures"))
  .orderBy($"run_date".desc)
  .show()

Why this matters:

Consistency: All pipelines use the same DQ framework
Centralized rules: Change validation logic without code changes
Auditability: DQ results stored in database for compliance
Fail-fast: Catch bad data before it reaches warehouse
Alerting: Automatic email on failures

Part 8 - Table operations: Backup, partitions, and location discovery

WHY table operations are hard

Spark makes it easy to query tables, but production pipelines need more:

Backup tables before risky operations
Drop specific partitions without dropping the whole table
Find table locations to copy files directly
Repair partition metadata after manual file operations

Standard Spark doesn’t provide these out of the box. You end up with:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
// Find location
val location = spark.sql("DESC FORMATTED my_table")
  .filter("col_name = 'Location'")
  .select("data_type")
  .first()
  .getAs[String]("data_type")

// Drop partitions
spark.sql("ALTER TABLE my_table DROP IF EXISTS PARTITION(date='2025-01-01') PURGE")

// Backup - what, copy files manually?

This is repetitive, error-prone, and not reusable.

Data loss risk

Always test backup utilities on non-production data first. The partition path extraction logic assumes standard Hive-style paths (warehouse.db/table/partition=value/). If your tables use custom locations or non-standard partitioning, you could silently copy files to wrong destinations or skip partitions entirely. Validate with a single partition before scaling up.

HOW to build table utilities

We’ll use the same pattern as before: implicit classes that extend DataFrame with table-specific operations.

The key insight: A DataFrame queried from a table “knows” which table it came from (via input file names). We can extract that metadata.

WHAT the backup utility looks like

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
implicit class ETLDataframeActions(dataframe: DataFrame) extends Serializable {

  def backup(backupLocationRootPath: String): Boolean = {
    val spark: SparkSession = SparkSession.getActiveSession.get
    val FILE_LOCATION = "file_location"
    val targetLocation = spark.sparkContext.broadcast(backupLocationRootPath)

    import spark.implicits.newStringEncoder

    // Step 1: Get all file locations from the DataFrame
    val fileLocations = dataframe
      .select(input_file_name().alias(FILE_LOCATION))
      .distinct()
      .map(_.getAs[String](FILE_LOCATION))
      .rdd
      .mapPartitions { partition =>
        partition.map { location =>
          // Step 2: Extract partition path from full location
          val locationTokens = location.split("/")
          val tablePath = locationTokens
            .slice(locationTokens.indexWhere(_.contains(".db")) + 2, locationTokens.length)
            .mkString("/")

          // Step 3: Build target path
          val targetPath = if (targetLocation.value.endsWith("/")) {
            targetLocation.value + tablePath
          } else {
            targetLocation.value + "/" + tablePath
          }

          (location, targetPath)
        }
      }

    // Step 4: Copy all files in parallel
    parallelCopy(fileLocations)
  }
}

How this works step-by-step:

1. Get source file locations:

input_file_name() is a Spark function that returns the full path of each input file
For a partitioned table: gs://bucket/warehouse.db/customers/date=2025-01-01/part-00000.parquet
.distinct() because multiple rows can come from the same file

2. Extract partition structure:

1
2
3
4
5
// Input: gs://bucket/warehouse.db/customers/date=2025-01-01/part-00000.parquet
// Split by /: ["gs:", "", "bucket", "warehouse.db", "customers", "date=2025-01-01", "part-00000.parquet"]
// Find .db: index 3
// Slice from .db+2 to end: ["date=2025-01-01", "part-00000.parquet"]
// Join: "date=2025-01-01/part-00000.parquet"

This preserves the partition structure in the backup.

flowchart TB
    A[DataFrame] --> B[input_file_name]
    B --> C[Extract File Locations]

    C --> D["gs://bucket/warehouse.db/<br/>customers/date=2025-01-01/<br/>part-00000.parquet"]

    D --> E[Split by '/']
    E --> F[Find '.db' index]
    F --> G[Slice from .db+2 to end]

    G --> H["date=2025-01-01/<br/>part-00000.parquet"]

    H --> I[Combine with backup path]

    I --> J["gs://backup-bucket/<br/>customers_backup/<br/>date=2025-01-01/<br/>part-00000.parquet"]

    J --> K[Parallel Copy<br/>via Spark RDD]

    style K fill:#90EE90,stroke:#2d7a2d,color:#000

3. Build target path:

1
2
// If target = "gs://backup/customers_backup/"
// Result = "gs://backup/customers_backup/date=2025-01-01/part-00000.parquet"

4. Parallel copy:

Uses Spark’s RDD parallelism to copy files across executors
Returns success/failure status

Usage:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
import com.vitthalmirji.dataengineering.spark.ETL._

val customersDf = spark.table("warehouse.customers")

// Backup entire table
val success = customersDf.backup("gs://backup-bucket/customers_backup/")

if (success) {
  println("Backup completed successfully")
} else {
  println("Backup failed - check logs")
}

// Backup specific partition
val partitionDf = spark.table("warehouse.customers").filter($"date" === "2025-06-28")
partitionDf.backup("gs://backup-bucket/customers_backup_20250628/")

Partition management

Drop partitions:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
implicit class ImplicitTablePartitionActions(tablePartitions: List[Partition]) extends Serializable {

  def drop(tableName: String): Unit = {
    val partitionList = getPartitions(inferType = true)
    val partitionString = partitionList
      .map(partition => s"PARTITION(${partition.mkString(",")})")
      .mkString(",")

    val query = s"ALTER TABLE $tableName DROP IF EXISTS $partitionString PURGE"
    logger.warn(s"Dropping partitions using query: $query")
    SparkSession.getActiveSession.foreach(_.sql(query).show())
  }

  def delete(tableName: String): Unit = {
    val partitionList = getPartitions(inferType = false)
    logger.warn(s"Attempting to delete partitions $partitionList")
    val tableLocation = getTableLocation(tableName)
    val dfsLocations = partitionList.map(partition =>
      tableLocation + "/" + partition.mkString("/")
    )
    dfsLocations.foreach(deleteDfsLocation)
  }
}

What’s a Partition?

1
2
3
4
5
case class Partition(partitions: Map[String, Any])

// Example
val part1 = Partition(Map("date" -> "2025-01-01"))
val part2 = Partition(Map("date" -> "2025-01-02", "region" -> "US"))

Usage:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
import com.vitthalmirji.dataengineering.spark.Table._

// Get partitions to drop
val targetTable = spark.table("warehouse.sales")
val partitionsToDrop = targetTable
  .filter($"date" < "2025-01-01")  // Old data
  .select("date", "region")
  .distinct()
  .collect()
  .map(row => Partition(Map(
    "date" -> row.getAs[String]("date"),
    "region" -> row.getAs[String]("region")
  )))
  .toList

// Drop metadata
partitionsToDrop.drop("warehouse.sales")

// Delete files
partitionsToDrop.delete("warehouse.sales")

// Repair metadata
repairRefreshTable("warehouse.sales")

Part 8.5 - GCS Object Operations: Copy, move, timestamps, and parallel operations

WHY blob operations need utilities

Working with GCS blobs in production pipelines requires common operations:

Copy files between buckets/locations
Move files (copy + delete source)
Get timestamps (created, last modified)
Check if blob is file vs directory
Parallel operations across 1000s of files

Standard approach:

1
2
3
4
5
6
7
8
val sourceBlob = storage.get("source-bucket", "file.parquet")
val copyWriter = sourceBlob.copyTo("target-bucket", "file.parquet")
val copyResult = copyWriter.getResult

if (copyResult.exists()) {
  sourceBlob.delete()
  println("Move complete")
}

Problems:

Verbose for every operation
No parallel execution built-in
Timestamp access requires Java time conversion
No “is this a file?” helper

HOW to wrap blob operations

We use implicit classes to extend Blob with utility methods:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
implicit class BlobOperations(blob: Blob) extends Serializable {
  def getValueAsString: String
  def createdTs(timeZone: Option[TimeZone]): LocalDateTime
  def updatedTs(timeZone: Option[TimeZone]): LocalDateTime
  def updatedDate(timeZone: Option[TimeZone]): LocalDate
  def isFile: Boolean
  def copy(targetBucket: String, targetObject: Option[String]): Boolean
  def move(targetBucket: String, targetObject: Option[String]): Unit
  def remove(): Boolean
}

WHAT the implementations look like

Get blob content as string:

1
def getValueAsString: String = blob.getContent().map(_.toChar).mkString

Timestamps with timezone support:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
def createdTs(timeZone: Option[TimeZone]): LocalDateTime =
  LocalDateTime.ofInstant(
    Instant.ofEpochMilli(blob.getCreateTime),
    timeZone.getOrElse(TZ_UTC).toZoneId
  )

def updatedTs(timeZone: Option[TimeZone]): LocalDateTime =
  LocalDateTime.ofInstant(
    Instant.ofEpochMilli(blob.getUpdateTime),
    timeZone.getOrElse(TZ_UTC).toZoneId
  )

def updatedDate(timeZone: Option[TimeZone]): LocalDate =
  updatedTs(timeZone).toLocalDate

Check if file (not directory):

1
2
3
4
5
def isFile: Boolean =
  trySafely(
    !blob.getName.endsWith("/"),
    Some(s"Error accessing blob ${blob.getName}")
  ).merge

Copy operation:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
def copy(
  targetBucketName: String,
  targetObjectName: Option[String],
  blobSourceOptions: BlobSourceOption*
): Boolean = {
  val copyResult = trySafely(
    if (targetObjectName.isDefined) {
      blob.copyTo(targetBucketName, targetObjectName.get, blobSourceOptions: _*)
    } else {
      blob.copyTo(targetBucketName, blobSourceOptions: _*)
    },
    Some(s"Error copying blob ${blob.getName}")
  )

  if (copyResult.merge.isDone) {
    true
  } else {
    logger.error(s"Blob doesn't exist ${blob.getName}")
    false
  }
}

Move operation (copy + delete):

1
2
3
4
5
6
7
8
def move(targetBucketName: String, targetObjectName: Option[String]): Unit = {
  val copyResult = copy(targetBucketName, targetObjectName)
  if (copyResult) {
    remove()
  } else {
    logger.warn(s"Source blob doesn't exist ${blob.getName}")
  }
}

Delete operation:

1
2
3
4
5
def remove(): Boolean =
  trySafely(
    blob.delete(),
    Some(s"Error deleting blob ${blob.getName}")
  ).merge

Parallel copy across RDD:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
def parallelCopy(fileLocations: RDD[(String, String)]): Boolean = {
  val copyResult = fileLocations.mapPartitions(partitionAttribute => {
    partitionAttribute.map { case (source, destination) =>
      logger.warn(s"Attempting to copy $source to $destination")
      val sourceBlob = blob"$source"
      val target = destination.toGcsObject
      val copyResult = sourceBlob.copy(target.bucketName, target.objectName)
      (sourceBlob, copyResult)
    }
  })

  val failedCopyBlobs = copyResult.filter(_._2.equals(false))
  val result = failedCopyBlobs.take(1).isEmpty

  if (!result) {
    logger.error("Error copying blobs: " +
      failedCopyBlobs.collect()
        .map(b => b._1.getBucket + "/" + b._1.getName)
        .mkString(","))
  }
  result
}

flowchart LR
    A[Source Blob] --> B{Operation}

    B -->|getValueAsString| C[Read Content<br/>as String]
    B -->|timestamps| D[createdTs<br/>updatedTs<br/>updatedDate]
    B -->|isFile| E[Check if File<br/>not directory]
    B -->|copy| F[Copy to Target]
    B -->|move| G[Copy + Delete Source]
    B -->|remove| H[Delete Blob]

    F --> I[Target Blob]
    G --> I

    style C fill:#87CEEB,stroke:#0066cc,color:#000
    style D fill:#87CEEB,stroke:#0066cc,color:#000
    style E fill:#87CEEB,stroke:#0066cc,color:#000
    style I fill:#90EE90,stroke:#2d7a2d,color:#000
    style H fill:#FFB6C6,stroke:#cc0066,color:#000

Usage in pipelines

Pattern 1: Backup files with timestamps:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
import com.vitthalmirji.dataengineering.gcp.GcsInterpolators._
import com.vitthalmirji.dataengineering.gcp.ObjectActions._

val sourceFile = blob"gs://source-bucket/data/important.parquet"
val lastModified = sourceFile.updatedDate(Some(tz"US/Central"))

// Backup with date suffix
val backupPath = s"gs://backup-bucket/data/important_$lastModified.parquet"
val backupTarget = backupPath.toGcsObject

val success = sourceFile.copy(backupTarget.bucketName, backupTarget.objectName)
if (success) {
  println(s"Backed up file from $lastModified")
}

Pattern 2: Move processed files to archive:

1
2
3
4
5
6
7
8
9
val processedFiles = blobs"gs://processing-bucket/output/"

processedFiles.getValues.asScala
  .filter(_.isFile)
  .foreach { blob =>
    val fileName = blob.getName.split("/").last
    blob.move("archive-bucket", Some(s"archive/2025/$fileName"))
    println(s"Archived: $fileName")
  }

Pattern 3: Clean up old files:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
val cutoffDate = yearMonthDate"2025-01-01"
val oldFiles = blobs"gs://temp-bucket/staging/"

oldFiles.getValues.asScala
  .filter(_.isFile)
  .filter(blob => blob.updatedDate(None).isBefore(cutoffDate))
  .foreach { blob =>
    blob.remove()
    println(s"Deleted old file: ${blob.getName}")
  }

Pattern 4: Read config from GCS:

1
2
3
4
5
6
7
8
val configBlob = blob"gs://config-bucket/pipeline-config.json"
val configJson = Some(configBlob.getValueAsString)

// Use with Converters
case class PipelineConfig(batchSize: Int, retries: Int)
val config = configJson.jsonStringAs[PipelineConfig]()

config.foreach(c => println(s"Batch size: ${c.batchSize}"))

Pattern 5: Parallel copy for backup (used by df.backup()):

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
import com.vitthalmirji.dataengineering.gcp.ObjectActions.parallelCopy

// Get file locations from DataFrame
val fileLocations: RDD[(String, String)] = df
  .select(input_file_name().alias("source"))
  .distinct()
  .rdd
  .map(row => {
    val source = row.getAs[String]("source")
    val target = source.replace("gs://prod-bucket/", "gs://backup-bucket/")
    (source, target)
  })

// Copy all files in parallel across executors
val success = parallelCopy(fileLocations)
if (success) {
  println("Parallel backup complete")
}

Part 8.7 - Affected partitions: 5 strategies for finding what changed

WHY affected partitions are critical

In production data lakes, you can’t reprocess everything every time. You need to know:

Which partitions changed since last run?
Which files were added/modified in the last 24 hours?
Which tables in a view have new data?

Without this, you either:

Reprocess everything (expensive, slow, wastes compute)
Manually track what changed (error-prone, doesn’t scale)

The challenge: Different use cases need different strategies.

The 5 strategies

flowchart TB
    A[Need Affected partitions?] --> B{Use Case}

    B -->|Simple: Single table<br/>Date partitions| C[Strategy 1:<br/>DataFrame-based<br/>input_file_name]

    B -->|Complex: Nested partitions<br/>Multi-level dirs| D[Strategy 2:<br/>GCS Multi-threaded<br/>Recursive traversal]

    B -->|Bucket-level:<br/>All files in bucket| E[Strategy 3:<br/>Bucket timestamp<br/>Parallel pages]

    B -->|File-level:<br/>Need file paths| F[Strategy 4:<br/>File-level GCS<br/>Returns paths]

    B -->|View-level:<br/>Multiple tables| G[Strategy 5:<br/>View with filters<br/>Per-table partitions]

    style C fill:#87CEEB,stroke:#0066cc,color:#000
    style D fill:#90EE90,stroke:#2d7a2d,color:#000
    style E fill:#FFD700,stroke:#cc8800,color:#000
    style F fill:#FFA500,stroke:#cc6600,color:#000
    style G fill:#FF69B4,stroke:#cc0066,color:#000

Strategy 1: DataFrame-based (Spark `input_file_name`)

WHEN to use:

Simple date-partitioned tables
Already have DataFrame loaded
Single-level partitions (e.g., date=2025-06-28)

HOW it works:

Use Spark’s input_file_name() to get file paths
Extract date from path using regex
Get blob from GCS and check updatedTs
Filter to files modified after since

WHAT the implementation looks like:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
implicit class TableDataframeActions(dataframe: DataFrame) extends Serializable {

  def getAffectedPartitionDates(
    since: LocalDateTime,
    dateRegex: String = "\\d{4}-\\d{2}-\\d{2}"
  ): Array[(LocalDate, LocalDateTime)] = {

    val FILE_LOCATION = "file_location"

    dataframe
      .select(input_file_name().alias(FILE_LOCATION))
      .distinct()
      .rdd
      .mapPartitions(partition =>
        partition.flatMap(field => {
          val location = field.getAs[String]("file_location")
          val blobLocation = blob"$location"
          val blobLastUpdated = blobLocation.updatedTs(Some(TZ_CST))

          if (blobLastUpdated.isAfter(since)) {
            Some((extractDate(location, dateRegex), blobLastUpdated))
          } else {
            None
          }
        })
      )
      .reduceByKey((v1, v2) => if (v1.isAfter(v2)) v1 else v2)
      .flatMap(t => t._1.map((_, t._2)))
      .distinct()
      .collect()
  }
}

Usage:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
import com.vitthalmirji.dataengineering.spark.Table._
import com.vitthalmirji.dataengineering.datetime.DateTimeInterpolators._

val customersDf = spark.table("warehouse.customers")
val cutoff = yearMonthDate24HrTs"2025-06-27 00:00:00"

val affectedDates = customersDf.getAffectedPartitionDates(since = cutoff)

affectedDates.foreach { case (date, lastModified) =>
  println(s"Partition $date was modified at $lastModified")

  // Reprocess only affected partition
  val partitionDf = customersDf.filter($"date" === date.toString)
  // ... process
}

Pros:

Fast for simple schemas
Uses Spark’s distributed processing
Works with any DataFrame

Cons:

Requires loading DataFrame first
Only works for tables Spark can read
Doesn’t handle nested partitions well

Strategy 2: GCS Multi-threaded (Recursive with Futures)

WHEN to use:

Complex nested partitions (e.g., date=2025-06-28/region=US/store=123/)
Need to traverse deep directory structures
Want maximum parallelism

HOW it works:

List GCS directory using blobs interpolator
For each page, spawn a Future (thread)
Recursively process subdirectories
Check file createdTs against since
Extract partition keys from path (date=X, region=Y)
Return Array[Option[Partition]]

WHAT the implementation looks like:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
def getAffectedPartitions(
  sourceTable: String,
  since: LocalDateTime
): Array[Option[Partition]] = {

  val tableLocation = getTableLocation(sourceTable)
  val blobLocation = if (tableLocation.endsWith("/")) tableLocation
                     else tableLocation.concat("/")

  val firstPage = blobs"$blobLocation"(
    Array(Storage.BlobListOption.currentDirectory())
  )

  val executionContext = ExecutionContext.fromExecutorService(
    Executors.newCachedThreadPool()
  )

  val futuresVector = GcsAffectedBlobs.processPages(
    firstPage,
    tableLocation.replace("gs://", "").concat("/"),
    Vector.empty,
    since,
    executionContext
  )

  val futures = Future.sequence(futuresVector)
  Await.result(futures, Duration.Inf)

  val affectedBlobs = futures.value.get.get.flatten.toArray.distinct

  executionContext.shutdown()
  affectedBlobs
}

The recursive multi-threaded logic:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
object GcsAffectedBlobs {

  @tailrec
  def processPages(
    currentPage: Page[Blob],
    parentBlobName: String,
    futures: Vector[Future[ListBuffer[Option[Partition]]]],
    since: LocalDateTime,
    executionContext: ExecutionContext
  ): Vector[Future[ListBuffer[Option[Partition]]]] =
    if (currentPage == null) {
      futures
    } else {
      processPages(
        currentPage.getNextPage,
        parentBlobName,
        futures ++ Vector(getAffectedBlobs(currentPage, parentBlobName, since, executionContext)),
        since,
        executionContext
      )
    }

  def getAffectedBlobs(
    page: Page[Blob],
    parentBlobName: String,
    since: LocalDateTime,
    executionContext: ExecutionContext
  ): Future[ListBuffer[Option[Partition]]] = Future {

    val affectedBlobsBuffer = ListBuffer.empty[Option[Partition]]
    val futuresBuffer = ListBuffer.empty[Future[ListBuffer[Option[Partition]]]]

    breakable {
      page.getValues.forEach { blob =>
        if (blob.getName.endsWith("/") && isValidDir(blob, parentBlobName)) {
          // Recurse into subdirectory
          val blobPath = blob.getBucket + "/" + blob.getName
          val blobs = blobs"$blobPath"(Array(Storage.BlobListOption.currentDirectory()))

          futuresBuffer.append(
            processPages(blobs, blobPath, Vector.empty, since, executionContext): _*
          )
        }
        else if (!blob.getName.endsWith("/") && blob.createdTs(None).isAfter(since)) {
          // Extract partition from path
          val partitionMap = blob.getName
            .split("/")
            .filter(_.contains("="))
            .flatMap { eachPartition =>
              val Array(k, v) = eachPartition.split("=")
              Array(k -> v)
            }
            .toMap

          if (partitionMap.nonEmpty) {
            affectedBlobsBuffer += Option(Partition(partitionMap))
            break  // Found affected file, stop checking this directory
          }
        }
      }
    }

    val futuresSeq = Future.sequence(futuresBuffer)
    Await.result(futuresSeq, Duration.Inf)

    futuresSeq.value.foreach(result =>
      result.foreach(list => list.flatten.foreach(partition => affectedBlobsBuffer += partition))
    )

    affectedBlobsBuffer.distinct
  }(executionContext)

  def isValidDir(blob: Blob, parentBlobName: String): Boolean = {
    val blobPath = blob.getBucket + "/" + blob.getName
    val blobName = blob.getName

    !blobPath.equalsIgnoreCase(parentBlobName) &&
    !blobName.contains("_spark_metadata") &&
    !blobName.contains("_staging") &&
    !blobName.contains(".hive-staging")
  }
}

Usage:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
import com.vitthalmirji.dataengineering.spark.Table._

val affectedPartitions = getAffectedPartitions(
  "warehouse.transactions",
  yearMonthDate24HrTs"2025-06-27 00:00:00"
)

affectedPartitions.foreach {
  case Some(Partition(partMap)) =>
    println(s"Affected partition: ${partMap.mkString(", ")}")

    // Build filter expression
    val filterExpr = partMap.map { case (k, v) => s"$k='$v'" }.mkString(" AND ")
    val df = spark.table("warehouse.transactions").filter(filterExpr)
    // ... reprocess
  case None =>
    println("No partitions affected")
}

Pros:

Handles deeply nested partitions
Maximum parallelism (one thread per page)
Early exit when affected file found in directory

Cons:

Complex implementation
Can create many threads
Requires careful Future management

Strategy 3: Bucket-level timestamps (Parallel pages)

WHEN to use:

Need ALL affected blobs in a bucket
Don’t care about partition structure
Want before/after timestamp filtering

HOW it works:

Get bucket from bucket interpolator
List all pages in parallel
Each page checked in separate thread
Filter by before/after timestamps with AND/OR
Return List[Blob]

WHAT the implementation looks like:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
def getAffectedBlobs(
  bucket: String,
  before: Option[LocalDateTime] = None,
  after: Option[LocalDateTime],
  operator: String = "OR"
): List[Blob] = {

  val gcsBucket = bucket"$bucket"
  var page = gcsBucket.list()

  val tasks = new util.ArrayList[ComputeAffectedBlobs]()
  do {
    tasks.add(new ComputeAffectedBlobs(page, before, after, operator))
    page = page.getNextPage
  } while (page != null)

  val workers: ExecutorService = Executors.newCachedThreadPool()
  val results: util.List[Future[ListBuffer[Blob]]] = workers.invokeAll(tasks)
  results.asScala.toList.flatMap(future => future.get())
}

class ComputeAffectedBlobs(
  page: Page[Blob],
  before: Option[LocalDateTime],
  after: Option[LocalDateTime],
  operator: String
) extends Callable[ListBuffer[Blob]] {

  override def call(): ListBuffer[Blob] = {
    val affectedBlobs = ListBuffer.empty[Blob]
    page.getValues.forEach { blob =>
      val blobLastUpdated = blob.updatedTs(None)
      (before, after, operator) match {
        case (Some(b), None, _) =>
          if (blobLastUpdated.isBefore(b)) affectedBlobs += blob
        case (None, Some(a), _) =>
          if (blobLastUpdated.isAfter(a)) affectedBlobs += blob
        case (Some(b), Some(a), "OR") =>
          if (blobLastUpdated.isBefore(b) || blobLastUpdated.isAfter(a))
            affectedBlobs += blob
        case (Some(b), Some(a), "AND") =>
          if (blobLastUpdated.isBefore(b) && blobLastUpdated.isAfter(a))
            affectedBlobs += blob
        case _ => None
      }
    }
    affectedBlobs
  }
}

Usage:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
// Get files modified in last 24 hours
val recentBlobs = getAffectedBlobs(
  bucket = "my-data-bucket",
  after = Some(yearMonthDate24HrTs"2025-06-27 00:00:00")
)

println(s"Found ${recentBlobs.size} recent files")

recentBlobs.foreach { blob =>
  println(s"${blob.getName} - ${blob.updatedTs(None)}")
}

// Get files in date range (between timestamps)
val rangeBlobs = getAffectedBlobs(
  bucket = "my-data-bucket",
  before = Some(yearMonthDate24HrTs"2025-06-28 00:00:00"),
  after = Some(yearMonthDate24HrTs"2025-06-27 00:00:00"),
  operator = "AND"
)

Pros:

Simple API (just bucket + timestamps)
Parallel page processing
Flexible timestamp filtering (before/after/AND/OR)

Cons:

Returns Blobs, not Partitions
No partition structure awareness
Processes entire bucket

Strategy 4: File-level GCS (Returns file paths)

WHEN to use:

Need exact file paths (not just partitions)
Want to process specific files
Building file-level lineage

HOW it works:

Similar to Strategy 2, but returns file paths
Recursively traverses directories
Returns Array[Option[String]] of GCS URIs

WHAT the implementation looks like:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
def getAffectedFiles(
  location: String,
  since: LocalDateTime
): Array[Option[String]] = {

  val blobLocation = if (location.endsWith("/")) location
                     else location.concat("/")

  val firstPage = blobs"$blobLocation"(
    Array(Storage.BlobListOption.currentDirectory())
  )

  val executionContext = ExecutionContext.fromExecutorService(Executors.newCachedThreadPool())

  val futuresVector = GcsAffectedBlobs.processAllPages(
    firstPage,
    location.replace("gs://", "").concat("/"),
    Vector.empty,
    since,
    executionContext
  )

  val futures = Future.sequence(futuresVector)
  Await.result(futures, Duration.Inf)

  val affectedFiles = futures.value.get.get.flatten.toArray.distinct

  executionContext.shutdown()
  affectedFiles
}

The file-level traversal:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
def getAllAffectedBlobs(
  page: Page[Blob],
  parentBlobName: String,
  since: LocalDateTime,
  executionContext: ExecutionContext
): Future[ListBuffer[Option[String]]] = Future {

  val affectedBlobsBuffer = ListBuffer.empty[Option[String]]
  val futuresBuffer = ListBuffer.empty[Future[ListBuffer[Option[String]]]]

  page.getValues.forEach { blob =>
    if (blob.getName.endsWith("/") && isValidDir(blob, parentBlobName)) {
      // Recurse
      val blobPath = blob.getBucket + "/" + blob.getName
      val blobs = blobs"$blobPath"(Array(Storage.BlobListOption.currentDirectory()))

      futuresBuffer.append(
        processAllPages(blobs, blobPath, Vector.empty, since, executionContext): _*
      )
    }
    else if (!blob.getName.endsWith("/") && blob.createdTs(None).isAfter(since)) {
      affectedBlobsBuffer += Some(s"gs://${blob.getBucket}/${blob.getName}")
    }
  }

  val futuresSeq = Future.sequence(futuresBuffer)
  Await.result(futuresSeq, Duration.Inf)

  futuresSeq.value.foreach(result =>
    result.foreach(list => list.flatten.foreach(filePath => affectedBlobsBuffer += filePath))
  )

  affectedBlobsBuffer.distinct
}(executionContext)

Usage:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
val affectedFiles = getAffectedFiles(
  "gs://my-bucket/warehouse/transactions/",
  yearMonthDate24HrTs"2025-06-27 00:00:00"
)

println(s"Found ${affectedFiles.length} affected files")

affectedFiles.foreach {
  case Some(filePath) =>
    println(s"Affected: $filePath")

    // Read specific file
    val df = spark.read.parquet(filePath)
    // ... process
  case None =>
    println("No files affected")
}

Pros:

Exact file paths returned
Can process individual files
Good for lineage tracking

Cons:

Returns paths, not partitions
Requires parsing paths to get partition info
Similar complexity to Strategy 2

Strategy 5: View-level with filters (Multi-table)

WHEN to use:

Working with views that union multiple tables
Need per-table partition lists
Have geo/region filters

HOW it works:

Parse view definition to find underlying tables
For each table, call Strategy 2
Apply optional filters (e.g., geo_region_cd IN ('US', 'MX'))
Return Array[MultiTablePartition]

WHAT the implementation looks like:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
case class MultiTablePartition(
  tableName: String,
  partitions: Array[Option[Partition]]
)

def getAffectedPartitions(
  sourceView: String,
  since: LocalDateTime,
  geoRegionFilters: Seq[String]
): Array[MultiTablePartition] = {

  val spark = SparkSession.getActiveSession.get
  val viewDf = spark.table(sourceView)

  // Extract tables from view
  val tableSeq = getScannedTableFromView(viewDf, geoRegionFilters)

  val tablesAffectedPartitions = tableSeq.map(table =>
    MultiTablePartition(table, getAffectedPartitions(table, since))
  )

  tablesAffectedPartitions.toArray
}

def getScannedTableFromView(
  viewDf: DataFrame,
  geoRegionFilters: Seq[String]
): Seq[String] = {
  // Parse query plan to extract table names
  // Apply filters if provided
  // Return list of table names
  ???  // Implementation details omitted for brevity
}

Usage:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
// View: warehouse.all_transactions (unions US, MX, CA tables)
val affectedTables = getAffectedPartitions(
  sourceView = "warehouse.all_transactions",
  since = yearMonthDate24HrTs"2025-06-27 00:00:00",
  geoRegionFilters = Seq("US", "MX")  // Only check US and MX tables
)

affectedTables.foreach { case MultiTablePartition(tableName, partitions) =>
  println(s"Table: $tableName")
  println(s"Affected partitions: ${partitions.length}")

  partitions.foreach {
    case Some(Partition(partMap)) =>
      println(s"  - ${partMap.mkString(", ")}")
    case None =>
  }

  // Reprocess each table's affected partitions
  partitions.foreach {
    case Some(Partition(partMap)) =>
      val filterExpr = partMap.map { case (k, v) => s"$k='$v'" }.mkString(" AND ")
      val df = spark.table(tableName).filter(filterExpr)
      // ... process
    case None =>
  }
}

Pros:

Handles complex view scenarios
Per-table partition tracking
Filter support for region/geo

Cons:

Most complex strategy
Requires view parsing
Multiple API calls (one per table)

Which strategy should you use?

flowchart TD
    A[Choose Strategy] --> B{Already have<br/>DataFrame loaded?}

    B -->|Yes| C{Simple<br/>partitions?}
    B -->|No| D{Need file paths<br/>or partitions?}

    C -->|Yes, date-only| E[Strategy 1:<br/>DataFrame-based]
    C -->|No, nested| F[Strategy 2:<br/>GCS Multi-threaded]

    D -->|Partitions| G{Nested<br/>partitions?}
    D -->|File paths| H[Strategy 4:<br/>File-level GCS]
    D -->|Just bucket| I[Strategy 3:<br/>Bucket-level]

    G -->|Yes| F
    G -->|No| J{Working with<br/>views?}

    J -->|Yes| K[Strategy 5:<br/>View-level]
    J -->|No| E

    style E fill:#87CEEB,stroke:#0066cc,color:#000
    style F fill:#90EE90,stroke:#2d7a2d,color:#000
    style H fill:#FFA500,stroke:#cc6600,color:#000
    style I fill:#FFD700,stroke:#cc8800,color:#000
    style K fill:#FF69B4,stroke:#cc0066,color:#000

Quick reference:

Strategy	Use Case	Returns	Performance	Complexity
1. DataFrame-based	Simple date partitions, already have DF	`Array[(LocalDate, LocalDateTime)]`	Fast	Low
2. GCS Multi-threaded	Nested partitions, max parallelism	`Array[Option[Partition]]`	Very Fast	High
3. Bucket-level	All files in bucket, timestamp range	`List[Blob]`	Fast	Medium
4. File-level GCS	Need exact file paths	`Array[Option[String]]`	Fast	High
5. View-level	Multiple tables, filters	`Array[MultiTablePartition]`	Medium	Very High

Production pattern: Combine strategies

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
def smartAffectedPartitions(
  tableName: String,
  since: LocalDateTime
): Array[Option[Partition]] = {

  val spark = SparkSession.getActiveSession.get

  // Try Strategy 1 first (fastest if table already loaded)
  val tryDf = Try(spark.table(tableName))

  tryDf match {
    case Success(df) =>
      // Check if simple partitioning
      val partitions = df.schema.filter(_.name.contains("date")).map(_.name)

      if (partitions.length == 1) {
        // Use DataFrame-based (Strategy 1)
        val dates = df.getAffectedPartitionDates(since)
        dates.map { case (date, _) =>
          Some(Partition(Map("date" -> date.toString)))
        }
      } else {
        // Fallback to GCS multi-threaded (Strategy 2)
        getAffectedPartitions(tableName, since)
      }

    case Failure(_) =>
      // Table not readable, use GCS (Strategy 2)
      getAffectedPartitions(tableName, since)
  }
}

Part 9 - Real production patterns: What I learned shipping this

Pattern 1: Composition over configuration

Bad:

1
2
3
def processData(input: String, output: String, mode: String, format: String, partitions: Int): Unit = {
  // 10 parameters, each adds complexity
}

Good:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
case class PipelineConfig(input: String, output: String, mode: SaveMode, format: String, partitions: Int)

def processData(config: PipelineConfig): Either[String, Unit] = {
  trySafely(
    unsafeCodeBlock = {
      val df = spark.read.format(config.format).load(config.input)
      // ... transform
      df.write.mode(config.mode).save(config.output)
    },
    errorMessage = Some(s"Pipeline failed for config: $config"),
    exceptionHandlingCodeblock = Some(Left(s"Failed: $config"))
  ) match {
    case Right(_) => Right(())
    case Left(err) => Left(err)
  }
}

Pattern 2: Always use trySafely for I/O

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
// Reading from external source
val sourceDf = trySafely(
  spark.read.parquet("gs://source-bucket/data/"),
  Some("Failed to read source"),
  Some(spark.emptyDataFrame)
) match {
  case Right(df) => df
  case Left(empty) =>
    logger.error("Source read failed, using empty DataFrame")
    empty
}

Pattern 3: CDC with audit trail

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
val deltaDf = sourceDf.performDelta(
  previousLoadedDataset = Some(targetDf),
  primaryKeys = List("id"),
  columnsExcludeHashcodeCalculation = Some(List("updated_at"))
)

// Write delta summary to audit table
val deltaAudit = deltaDf.groupBy("delta_flag").count()
  .withColumn("pipeline_run_ts", current_timestamp())
  .withColumn("table_name", lit("customers"))

deltaAudit.write.mode("append").saveAsTable("audit.cdc_summary")

Pattern 4: Complete table maintenance workflow

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
def maintainTable(
  tableName: String,
  zOrderCols: Seq[String],
  retentionHours: Double = 168.0
): Unit = {
  import com.vitthalmirji.dataengineering.spark.DeltaTableUtils._

  logger.info(s"Starting maintenance for $tableName")

  // Step 1: Compact small files
  if (!executeCompaction(tableName)) {
    throw new RuntimeException(s"Compaction failed")
  }

  // Step 2: Z-order for query performance
  if (!executeZOrderBy(tableName, Some(zOrderCols))) {
    throw new RuntimeException(s"Z-ordering failed")
  }

  // Step 3: Vacuum old files
  if (!executeVacuum(tableName, retentionHours)) {
    logger.warn("Vacuum failed - old files remain")
  }

  logger.info(s"Maintenance complete for $tableName")
}

// Usage - run nightly
maintainTable("warehouse.transactions", Seq("customer_id", "date"))

Pattern 5: Schema evolution with DQ checks

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
import com.vitthalmirji.dataengineering.schema.SchemaOps._
import com.vitthalmirji.dataengineering.spark.DataQuality._

val sourceDf = spark.table("source.customers")
val targetTable = "warehouse.customers"
val targetDf = spark.table(targetTable)

// Detect schema changes
if (isSchemaChanged(sourceDf, targetDf)) {
  logger.warn("Schema change detected - evolving")

  val input = SchemaEvolution(
    sourceDf = sourceDf,
    targetTable = targetTable,
    primaryKeys = Seq("customer_id")
  )

  evolveNonPartitionedTable(input)
}

// Always run DQ after evolution
val isValid = spark.table(targetTable).executeDQ(ruleSetId = 200)

if (!isValid) {
  throw new RuntimeException("DQ failed post-evolution")
}

Pattern 6: Type-safe date filtering

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
import com.vitthalmirji.dataengineering.datetime.DateTimeInterpolators._

// Instead of string dates (error-prone)
val badDf = df.filter($"date" >= "2025-06-28")  // Typo: "2025-06-32" would pass

// Use interpolators (fails fast)
val startDate = yearMonthDate"2025-06-01"
val endDate = yearMonthDate"2025-06-28"

val goodDf = df.filter(
  $"date" >= lit(startDate.toString) && $"date" <= lit(endDate.toString)
)

Appendix A - Complete Pipeline Architecture

Here’s how all utilities work together in a production pipeline:

graph TB
    subgraph "Data Sources"
        S1[Source Tables]
        S2[Config JSON]
        S3[External APIs]
    end

    subgraph "Utilities Layer"
        U1[trySafely<br/>Error Handling]
        U2[Converters<br/>JSON/Java↔Scala]
        U3[Date Interpolators<br/>Type-Safe Dates]
        U4[Schema Ops<br/>Detect Changes]
    end

    subgraph "Pipeline Logic"
        P1[Read Data]
        P2{Schema Changed?}
        P3[Schema Evolution]
        P4[Transform Data]
        P5[CDC performDelta]
        P6{DQ Check}
    end

    subgraph "Delta Lake Ops"
        D1[Compact]
        D2[Z-Order]
        D3[Vacuum]
    end

    subgraph "Output"
        O1[Warehouse Tables]
        O2[Staging Tables]
        O3[Audit DB]
    end

    S1 --> U1
    S2 --> U2
    S3 --> U1
    U1 --> P1
    U2 --> P1
    U3 --> P1

    P1 --> P2
    P2 -->|Yes| P3
    P2 -->|No| P4
    P3 --> P4
    P4 --> U4
    U4 --> P5
    P5 --> P6

    P6 -->|Pass| O1
    P6 -->|Fail| O2
    P6 --> O3

    O1 --> D1
    D1 --> D2
    D2 --> D3

    style P6 fill:#FFD700,stroke:#cc8800,color:#000
    style O1 fill:#90EE90,stroke:#2d7a2d,color:#000
    style O2 fill:#FFB6C6,stroke:#cc0066,color:#000
    style D3 fill:#87CEEB,stroke:#0066cc,color:#000

Utility Class Relationships

classDiagram
    class Helpers {
        +trySafely[T,E,C](unsafeCodeBlock, errorMessage, fallback) Either[C,T]
        +handleException[E,C](exception, message, fallback) C
    }

    class Converters {
        <<implicit class>>
        +JavaToScalaOption.toOption() Option[T]
        +JavaToScalaList.toScala() List[E]
        +JavaToScalaMap.toScala() Map[K,V]
        +VariousTypesFromJsonString.jsonStringAs[T]() Option[T]
    }

    class DateTimeInterpolators {
        <<implicit class>>
        +yearMonthDate() LocalDate
        +yearMonthDate24HrTs() LocalDateTime
        +tz() TimeZone
    }

    class ETL {
        <<implicit class>>
        +DeltaTransformations.performDelta() DataFrame
        +ETLDataframeActions.backup() Boolean
    }

    class DeltaTableUtils {
        +executeVacuum(tableName, retentionHours) Boolean
        +executeCompaction(tableName, partitionFilter) Boolean
        +executeZOrderBy(tableName, keys, partitionFilter) Boolean
    }

    class SchemaOps {
        +isSchemaChanged(df1, df2) Boolean
        +getSchemaMap(df) Map[String,String]
        +newColumnsAdded(oldDf, newDf) Boolean
        +getNewColumnsDf(oldDf, newDf, primaryKeys) DataFrame
    }

    class DataQuality {
        <<implicit class>>
        +DataQualityActions.executeDQ(ruleSetId, email) Boolean
        +executeDQWithDBWrite(ruleSetId, dbProps) Boolean
    }

    class Table {
        <<implicit class>>
        +ImplicitTablePartitionActions.drop(tableName) Unit
        +ImplicitTablePartitionActions.delete(tableName) Unit
    }

    ETL ..> Helpers : uses
    DataQuality ..> Helpers : uses
    SchemaOps ..> Helpers : uses
    ETL ..> DateTimeInterpolators : uses
    ETL ..> DeltaTableUtils : uses
    DataQuality ..> SchemaOps : uses

Appendix B - Full project structure

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
dataengineering-savvy/
├── build.sbt
└── src/
    ├── main/scala/com/vitthalmirji/dataengineering/
    │   ├── common/
    │   │   ├── Helpers.scala              # trySafely, error handling
    │   │   ├── Converters.scala           # Java↔Scala, JSON parsing
    │   │   └── Email.scala                # Email validation utilities
    │   ├── datetime/
    │   │   ├── DateTimeInterpolators.scala # yearMonthDate, tz interpolators
    │   │   ├── DateTimeHelpers.scala       # Date extraction, formatting
    │   │   └── DateTimeConstants.scala     # Format constants
    │   ├── spark/
    │   │   ├── ETL.scala                    # performDelta, backup, audit columns
    │   │   ├── Table.scala                  # Partition ops, table location
    │   │   ├── Partition.scala              # Partition representation
    │   │   ├── DeltaTableUtils.scala        # Vacuum, compaction, Z-ordering
    │   │   └── DataQuality.scala            # DQ wrapper utilities
    │   ├── schema/
    │   │   └── SchemaOps.scala              # Schema comparison, evolution
    │   ├── gcp/
    │   │   ├── GcsInterpolators.scala       # blob"gs://..." interpolator
    │   │   ├── ObjectActions.scala          # GCS copy, delete operations
    │   │   └── GcsObject.scala              # GCS metadata wrapper
    │   └── constants/
    │       ├── StringConstants.scala        # Audit column names, delimiters
    │       └── DateTimeConstants.scala      # Date format patterns
    └── test/scala/com/vitthalmirji/dataengineering/
        ├── common/
        │   ├── HelpersTest.scala
        │   └── ConvertersTest.scala
        ├── datetime/
        │   └── DateTimeInterpolatorsTest.scala
        ├── spark/
        │   ├── ETLTest.scala                # CDC tests
        │   ├── TableTest.scala              # Partition tests
        │   ├── DeltaTableUtilsTest.scala    # Delta ops tests
        │   └── DataQualityTest.scala        # DQ tests
        └── schema/
            └── SchemaOpsTest.scala          # Schema comparison tests

build.sbt:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
ThisBuild / scalaVersion := "2.13.12"
ThisBuild / organization := "com.vitthalmirji"

lazy val root = (project in file("."))
  .settings(
    name := "dataengineering-savvy",
    libraryDependencies ++= Seq(
      "org.apache.spark" %% "spark-sql" % "3.5.0" % "provided",
      "io.delta" %% "delta-core" % "2.4.0" % "provided",
      "com.fasterxml.jackson.module" %% "jackson-module-scala" % "2.15.2",
      "org.scalatest" %% "scalatest" % "3.2.17" % Test
    )
  )

Clone and run:

1
2
3
4
git clone https://github.com/vim89/dataengineering-savvy.git
cd dataengineering-savvy
sbt compile
sbt test

Example usage in your pipeline:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
// build.sbt
libraryDependencies += "com.vitthalmirji" %% "dataengineering-savvy" % "1.0.0"

// Pipeline.scala
import com.vitthalmirji.dataengineering.common.Helpers._
import com.vitthalmirji.dataengineering.datetime.DateTimeInterpolators._
import com.vitthalmirji.dataengineering.spark.ETL._
import com.vitthalmirji.dataengineering.spark.DeltaTableUtils._
import com.vitthalmirji.dataengineering.spark.DataQuality._
import com.vitthalmirji.dataengineering.schema.SchemaOps._

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

    val runDate = yearMonthDate"2025-06-28"

    // Read with error handling
    val sourceDf = trySafely(
      spark.read.parquet(s"gs://source/data/date=$runDate"),
      Some("Failed to read source"),
      Some(spark.emptyDataFrame)
    ).getOrElse(spark.emptyDataFrame)

    // Check schema evolution
    val targetTable = "warehouse.customers"
    if (isSchemaChanged(sourceDf, spark.table(targetTable))) {
      logger.warn("Schema evolved - handling automatically")
      evolveNonPartitionedTable(SchemaEvolution(
        sourceDf, targetTable, Seq("customer_id")
      ))
    }

    // Run DQ checks
    val isValid = sourceDf.executeDQ(ruleSetId = 200)

    if (isValid) {
      // Perform CDC
      val deltaDf = sourceDf.performDelta(
        Some(spark.table(targetTable)),
        List("customer_id")
      )

      // Write
      deltaDf.write.mode("overwrite").insertInto(targetTable)

      // Maintain table
      executeCompaction(targetTable)
      executeZOrderBy(targetTable, Some(Seq("customer_id", "region")))
      executeVacuum(targetTable, retentionHours = 168.0)
    } else {
      throw new RuntimeException("DQ failed - NOT writing to warehouse")
    }
  }
}

References

Scala Functional Programming

Functional Programming in Scala (Red Book) - Deep dive into functional patterns
Scala with Cats - Type classes and abstractions
Scala Implicits Guide - Official docs on implicit classes

Apache Spark & Delta Lake

Spark SQL Programming Guide - DataFrame operations
Spark Data Sources - Reading/writing data
Delta Lake Documentation - Delta Lake operations and best practices
Delta Lake Optimization Guide - Compaction, Z-ordering, vacuum

Data Engineering Patterns

The Data Warehouse Toolkit - SCD patterns and dimensional modeling
Building the Data Lakehouse - Modern data architecture patterns

Open Source

dataengineering-savvy on GitHub - Full source code with tests

TL;DR

Higher-order functions (trySafely) wrap unsafe operations generically - write once, use everywhere
Implicit classes extend types without inheritance - add .toScala, .backup(), .executeDQ() to existing types
String interpolators (yearMonthDate"...") make domain-specific syntax readable and type-safe
CDC with hash-based comparison is performant and deterministic for SCD Type 1
Delta Lake utilities (vacuum, compaction, Z-ordering) automate table maintenance
Schema evolution detects and handles column adds/drops with validation
Data quality wrappers integrate DQ frameworks seamlessly into pipelines
Schema comparison utilities catch schema drift early
Table operations (backup, partition drop, location discovery) are production necessities
Functional utilities reduce boilerplate by 70%+ and make pipelines maintainable
Type safety prevents entire classes of runtime errors - if it compiles, error handling is consistent
The real value: your team uses the SAME patterns, not 10 different ways to parse a date or vacuum a table

Vitthal Mirji

Staff Data Engineer @ Walmart

Mumbai, India

Staff Data Engineer & Architect from Mumbai, India. Sharing insights on Data Engineering, Functional programming, Scala, Open source, and life.

Expertise

Data Engineering
Scala
Apache Spark
Functional Programming
Cloud Architecture
GCP
Big Data

Edit this on GitHub

Next time, we'll talk about "Currying: Not Just for Food, Also for Making Simple Things Complicated"

Functional utilities for data engineering: stop repeats in Scala pipelines now

Why functional utilities matter

What we’ll build

Part 1 - Higher-order functions: Wrapping unsafe operations

WHY we need this

HOW it works: Higher-order functions

WHAT the implementation looks like

Usage examples

Why this pattern works

Part 2 - Implicit classes: Extending types without inheritance

WHY we need implicit conversions

HOW implicit classes work

WHAT the converters look like

The converter suite

JSON parsing with implicits

Part 3 - String interpolators: Making date parsing readable

WHY dates are painful

HOW string interpolators work

WHAT the date interpolator looks like

Usage in pipelines

Why this matters

Part 3.5 - GCS String Interpolators: Working with Cloud Storage like a pro

WHY GCS operations are painful

HOW GCS interpolators work

WHAT the complete GCS interpolator suite looks like

Usage in pipelines

Why this matters

Part 4 - Change Data Capture (CDC): SCD Type 1 with Spark

WHY CDC matters

HOW Delta CDC works

WHAT the implementation looks like

Usage example

Part 5 - Delta Lake utilities: Vacuum, compaction, and Z-ordering

WHY Delta Lake needs maintenance

HOW to wrap Delta ops

WHAT the Delta utilities look like

Usage in pipelines

Part 6 - Schema evolution: Automatic column add/drop with validation

WHY schema evolution is hard

HOW to automate schema evolution

WHAT the schema comparison utilities look like

WHAT the schema evolution workflow looks like

Usage in pipelines

Part 7 - Data quality wrappers: Integrating DQ frameworks

WHY data quality checks matter

HOW to wrap DQ frameworks

WHAT the DQ wrapper looks like

Usage in pipelines

Part 8 - Table operations: Backup, partitions, and location discovery

WHY table operations are hard

HOW to build table utilities

WHAT the backup utility looks like

Partition management

Part 8.5 - GCS Object Operations: Copy, move, timestamps, and parallel operations

WHY blob operations need utilities

HOW to wrap blob operations

WHAT the implementations look like

Usage in pipelines

Part 8.7 - Affected partitions: 5 strategies for finding what changed

WHY affected partitions are critical

The 5 strategies

Strategy 1: DataFrame-based (Spark input_file_name)

Strategy 2: GCS Multi-threaded (Recursive with Futures)

Strategy 3: Bucket-level timestamps (Parallel pages)

Strategy 4: File-level GCS (Returns file paths)

Strategy 5: View-level with filters (Multi-table)

Which strategy should you use?

Production pattern: Combine strategies

Part 9 - Real production patterns: What I learned shipping this

Pattern 1: Composition over configuration

Pattern 2: Always use trySafely for I/O

Pattern 3: CDC with audit trail

Pattern 4: Complete table maintenance workflow

Pattern 5: Schema evolution with DQ checks

Pattern 6: Type-safe date filtering

Appendix A - Complete Pipeline Architecture

Utility Class Relationships

Appendix B - Full project structure

References

Scala Functional Programming

Strategy 1: DataFrame-based (Spark `input_file_name`)