Apache Spark Provider Guide¶

This guide covers how to construct and run pipelines using the Apache Spark providers — the first concrete implementation in PatternOps.

Available Spark Providers¶

Provider	Class	Capability
Spark File Source	`SparkFileSourceProvider`	Source Acquisition (file-based)
Spark JDBC Source	`SparkJdbcSourceProvider`	Source Acquisition (database)
Spark Transformation	`SparkTransformationProvider`	Transformation (SQL + DataFrame)
Spark File Storage	`SparkFileStorageProvider`	Storage (file output)
Spark JDBC Storage	`SparkJdbcStorageProvider`	Storage (database output)

File Source Configuration¶

Read data from CSV, Parquet, JSON, ORC, Avro, or Delta Lake files.

Configuration Keys¶

Key	Required	Default	Description
`path`	✅	—	File or directory path
`format`	✅	—	`csv`, `parquet`, `json`, `orc`, `avro`, `delta`
`header`	❌	`true`	CSV: include header row
`inferSchema`	❌	`true`	CSV: infer column types
`delimiter`	❌	`,`	CSV: field delimiter
`multiLine`	❌	`false`	JSON: multi-line JSON objects
`mergeSchema`	❌	`false`	Parquet/Delta: merge schemas across files

Acquisition Modes¶

Mode	Behaviour
`snapshot`	Read all files at the path (full extract)
`incremental`	Read only files modified after the last checkpoint

Example: CSV Source¶

- name: acquire-customers
  type: source
  capability: source-acquisition
  provider: spark-file-source
  timeout: PT5M
  config:
    path: /data/input/customers.csv
    format: csv
    header: true
    inferSchema: true
    delimiter: ","

Example: Parquet Source (Directory)¶

- name: acquire-events
  type: source
  capability: source-acquisition
  provider: spark-file-source
  timeout: PT10M
  config:
    path: /data/lake/events/
    format: parquet
    mergeSchema: true

Example: Delta Lake Source¶

- name: acquire-delta
  type: source
  capability: source-acquisition
  provider: spark-file-source
  timeout: PT5M
  config:
    path: /data/lake/customers_delta/
    format: delta

JDBC Source Configuration¶

Read data from relational databases via Spark's JDBC connector.

Configuration Keys¶

Key	Required	Default	Description
`url`	✅	—	JDBC connection URL
`table`	✅	—	Table name or subquery
`driver`	❌	auto-detect	JDBC driver class
`user`	❌	—	Database username
`password`	❌	—	Database password
`connectionTimeout`	❌	`30`	Connection timeout (seconds, max 300)
`fetchSize`	❌	`1000`	JDBC fetch size
`partitionColumn`	❌	—	Column for parallel reads
`lowerBound`	❌	—	Lower bound for partition column
`upperBound`	❌	—	Upper bound for partition column
`numPartitions`	❌	`4`	Number of read partitions
`watermarkColumn`	❌	—	Column for incremental reads
`watermarkValue`	❌	—	Last watermark value

Acquisition Modes¶

Mode	Behaviour
`snapshot`	Full table read
`incremental`	WHERE `watermarkColumn > watermarkValue`

Example: Full Table Read¶

- name: acquire-orders
  type: source
  capability: source-acquisition
  provider: spark-jdbc-source
  timeout: PT10M
  config:
    url: jdbc:postgresql://db-host:5432/orders
    table: customer_orders
    driver: org.postgresql.Driver
    user: reader
    password: ${DB_PASSWORD}

Example: Partitioned Read (Large Tables)¶

- name: acquire-large-table
  type: source
  capability: source-acquisition
  provider: spark-jdbc-source
  timeout: PT30M
  config:
    url: jdbc:postgresql://db-host:5432/warehouse
    table: transactions
    driver: org.postgresql.Driver
    partitionColumn: id
    lowerBound: 1
    upperBound: 10000000
    numPartitions: 16
    fetchSize: 5000

Example: Incremental Load with Watermark¶

- name: acquire-incremental
  type: source
  capability: source-acquisition
  provider: spark-jdbc-source
  timeout: PT10M
  config:
    url: jdbc:postgresql://db-host:5432/orders
    table: customer_orders
    driver: org.postgresql.Driver
    watermarkColumn: updated_at
    watermarkValue: "2024-01-15 00:00:00"

Transformation Configuration¶

Transform data using SQL or DataFrame API.

Transformation Styles¶

Style	Description	Reference Format
`sql`	SQL query against a temp view	SQL string with `{input}` placeholder
`dataframe`	DataFrame operations (SQL-based path)	SQL string with `{input}` placeholder

The `{input}` Placeholder¶

In SQL references, {input} is replaced with the temp view name from the previous stage's output:

SELECT id, name, UPPER(city) AS city, age 
FROM {input} 
WHERE age >= 18

Example: Filter and Project¶

- name: filter-active-customers
  type: transformation
  capability: transformation
  provider: spark-transformation
  timeout: PT10M
  transformation:
    sql:
      style: sql
      reference: "SELECT id, name, email, city FROM {input} WHERE status = 'active'"

Example: Aggregation¶

- name: aggregate-orders
  type: transformation
  capability: transformation
  provider: spark-transformation
  timeout: PT15M
  transformation:
    sql:
      style: sql
      reference: |
        SELECT 
          customer_id,
          SUM(amount) AS total_amount,
          COUNT(*) AS order_count,
          MAX(order_date) AS last_order_date
        FROM {input}
        GROUP BY customer_id

Example: Join (Multiple Sources)¶

- name: enrich-with-lookup
  type: transformation
  capability: transformation
  provider: spark-transformation
  timeout: PT10M
  transformation:
    sql:
      style: sql
      reference: |
        SELECT o.*, c.name AS customer_name, c.city
        FROM {input} o
        JOIN customer_lookup c ON o.customer_id = c.id

Example: Window Functions¶

- name: rank-customers
  type: transformation
  capability: transformation
  provider: spark-transformation
  timeout: PT10M
  transformation:
    sql:
      style: sql
      reference: |
        SELECT *,
          ROW_NUMBER() OVER (PARTITION BY city ORDER BY total_amount DESC) AS city_rank
        FROM {input}

Example: Dual-Mode (Spark + SQL)¶

- name: complex-transform
  type: transformation
  capability: transformation
  provider: spark-transformation
  timeout: PT15M
  transformation:
    spark:
      style: dataframe
      reference: com.acme.transforms.ComplexEnrichment
    sql:
      style: sql
      reference: transforms/complex_enrichment.sql

File Storage Configuration¶

Write data to file-based storage.

Write Options¶

Key	Required	Default	Description
`format`	❌	`parquet`	Output format: `parquet`, `delta`, `csv`, `json`
`mode`	❌	`overwrite`	Write mode: `overwrite`, `append`, `ignore`, `errorifexists`
`partitionBy`	❌	—	Columns to partition by (comma-separated or list)
`header`	❌	`true`	CSV: include header row

Example: Write Parquet with Partitioning¶

- name: write-output
  type: publish
  capability: storage
  provider: spark-file-storage
  timeout: PT5M
  config:
    format: parquet
    mode: overwrite
    partitionBy: year,month

Example: Write Delta Lake (Append)¶

- name: write-delta
  type: publish
  capability: storage
  provider: spark-file-storage
  timeout: PT5M
  config:
    format: delta
    mode: append
    partitionBy: processed_date

Example: Write CSV¶

- name: export-csv
  type: publish
  capability: storage
  provider: spark-file-storage
  timeout: PT5M
  config:
    format: csv
    mode: overwrite
    header: true

JDBC Storage Configuration¶

Write data to relational databases.

Write Options¶

Key	Required	Default	Description
`table`	✅	—	Target table name
`mode`	❌	`append`	Write mode: `overwrite`, `append`
`batchSize`	❌	`1000`	Batch size for inserts

Example: Write to Database¶

- name: write-to-analytics
  type: publish
  capability: storage
  provider: spark-jdbc-storage
  timeout: PT10M
  config:
    url: jdbc:postgresql://analytics-db:5432/warehouse
    table: customer_summary
    driver: org.postgresql.Driver
    mode: overwrite
    batchSize: 5000

Running Pipelines Programmatically¶

import io.patternops.providers.spark.runner.PipelineRunner;
import io.patternops.providers.spark.runner.PipelineExecutionResult;
import io.patternops.core.parser.ConfigurationParser;
import io.patternops.core.registry.DefaultProviderRegistry;
import io.patternops.core.echo.DefaultEchoService;
import io.patternops.core.lifecycle.StageLifecycleExecutor;

// Create dependencies
ConfigurationParser parser = new ConfigurationParser();
DefaultProviderRegistry registry = new DefaultProviderRegistry();
DefaultEchoService echoService = new DefaultEchoService();
StageLifecycleExecutor lifecycleExecutor = new StageLifecycleExecutor();

// Create runner
PipelineRunner runner = new PipelineRunner(parser, registry, echoService, lifecycleExecutor);

// Run from YAML
String yaml = Files.readString(Path.of("my-pipeline.yaml"));
PipelineExecutionResult result = runner.run(yaml);

// Check result
if (result.isSuccess()) {
    System.out.println("Pipeline completed in " + result.totalDuration());
    System.out.println("Stages completed: " + result.completedStageCount());
} else {
    System.err.println("Pipeline failed:");
    result.errors().forEach(e -> System.err.println("  - " + e));
}

Common Patterns¶

Pattern: Incremental Load with Checkpoint¶

stages:
  - name: acquire-incremental
    type: source
    capability: source-acquisition
    provider: spark-jdbc-source
    timeout: PT10M
    config:
      url: jdbc:postgresql://source:5432/db
      table: orders
      watermarkColumn: updated_at
      watermarkValue: "2024-01-01 00:00:00"

The watermarkValue is typically stored externally (e.g., in the Echo service state) and updated after each successful run.

Pattern: Multi-Stage Transformation¶

stages:
  - name: clean-data
    type: transformation
    capability: transformation
    provider: spark-transformation
    timeout: PT5M
    transformation:
      sql:
        style: sql
        reference: "SELECT * FROM {input} WHERE id IS NOT NULL AND amount > 0"

  - name: enrich-data
    type: transformation
    capability: transformation
    provider: spark-transformation
    timeout: PT10M
    transformation:
      sql:
        style: sql
        reference: |
          SELECT t.*, l.region, l.country
          FROM {input} t
          JOIN location_lookup l ON t.city = l.city

Pattern: Write to Multiple Targets¶

stages:
  - name: write-parquet
    type: publish
    capability: storage
    provider: spark-file-storage
    timeout: PT5M
    config:
      format: parquet
      mode: overwrite

  - name: write-to-db
    type: publish
    capability: storage
    provider: spark-jdbc-storage
    timeout: PT10M
    config:
      url: jdbc:postgresql://analytics:5432/warehouse
      table: summary_table
      mode: overwrite

Troubleshooting¶

Common Issues¶

Issue	Cause	Solution
`ClassNotFoundException: org.postgresql.Driver`	JDBC driver not on classpath	Add driver JAR to Spark's `--jars` or `extraClassPath`
`Path does not exist`	Invalid file path	Verify path exists and is readable
`Connection timed out`	Database unreachable	Check network, increase `connectionTimeout`
`Schema mismatch`	Parquet files with different schemas	Set `mergeSchema: true`
`AnalysisException: Table not found`	Temp view not registered	Ensure previous stage completed successfully

Debugging Tips¶

Check schema discovery first: Use discoverSchema() to verify the source is readable
Use explain(): The transformation provider's explain method shows the Spark query plan
Check Echo metrics: Stage duration metrics help identify slow stages
Use test mode: Run with sample data before full production loads

Apache Spark Provider Guide¶

Available Spark Providers¶

File Source Configuration¶

Configuration Keys¶

Acquisition Modes¶

Example: CSV Source¶

Example: Parquet Source (Directory)¶

Example: Delta Lake Source¶

JDBC Source Configuration¶

Configuration Keys¶

Acquisition Modes¶

Example: Full Table Read¶

Example: Partitioned Read (Large Tables)¶

Example: Incremental Load with Watermark¶

Transformation Configuration¶

Transformation Styles¶

The {input} Placeholder¶

Example: Filter and Project¶

Example: Aggregation¶

Example: Join (Multiple Sources)¶

Example: Window Functions¶

Example: Dual-Mode (Spark + SQL)¶

File Storage Configuration¶

Write Options¶

Example: Write Parquet with Partitioning¶

Example: Write Delta Lake (Append)¶

Example: Write CSV¶

JDBC Storage Configuration¶

Write Options¶

Example: Write to Database¶

Running Pipelines Programmatically¶

Common Patterns¶

Pattern: Incremental Load with Checkpoint¶

Pattern: Multi-Stage Transformation¶

Pattern: Write to Multiple Targets¶

Troubleshooting¶

Common Issues¶

Debugging Tips¶

The `{input}` Placeholder¶