Skip to content

Quick Start

This guide walks you through defining, parsing, validating, and printing your first PatternOps pipeline.

Step 1: Define a Pipeline

Create a file my-pipeline.yaml:

name: customer-ingestion
tenancy: acme-corp
namespace: production
dataset: customers
executionMode: batch
platformProfile: databricks

stages:
  - name: acquire-customers
    type: source
    capability: source-acquisition
    provider: jdbc-provider
    timeout: PT5M
    config:
      mode: incremental
      watermark: updated_at

  - name: validate-data
    type: quality
    capability: data-quality
    timeout: PT3M

  - name: transform-customers
    type: transformation
    capability: transformation
    timeout: PT10M
    retryPolicy:
      maxAttempts: 3
      backoff: PT10S
      strategy: exponential
    transformation:
      spark:
        style: dataframe
        reference: com.acme.transforms.CustomerEnrichment
      sql:
        style: sql
        reference: transforms/customer_enrichment.sql

  - name: publish-output
    type: publish
    capability: storage
    timeout: PT5M

policies: []
extensions: []

Step 2: Parse the Pipeline

import io.patternops.core.DefaultPipelineDefinitionModel;
import io.patternops.core.model.Pipeline;
import io.patternops.core.service.PipelineDefinitionModel;
import io.patternops.core.service.PipelineDefinitionModel.ValidationResult;

// Create the model
PipelineDefinitionModel model = new DefaultPipelineDefinitionModel();

// Read your YAML file
String yaml = Files.readString(Path.of("my-pipeline.yaml"));

// Parse
ValidationResult<Pipeline> result = model.parse(yaml);

if (result.valid()) {
    Pipeline pipeline = result.value();
    System.out.println("Pipeline: " + pipeline.name());
    System.out.println("Stages: " + pipeline.stages().size());
    System.out.println("Mode: " + pipeline.executionMode());
} else {
    System.err.println("Validation errors:");
    result.violations().forEach(v -> System.err.println("  - " + v));
}

Step 3: Validate the Pipeline

// Validate against schema rules (provider-specific construct detection, etc.)
ValidationResult<Pipeline> validation = model.validate(pipeline);

if (!validation.valid()) {
    System.err.println("Schema violations:");
    validation.violations().forEach(v -> System.err.println("  - " + v));
    // Example output:
    // - pipeline.stages[0].config.sparkConf: contains provider-specific construct (Spark-specific configuration)
}

Step 4: Resolve Bindings

import io.patternops.core.service.ProviderRegistry;

// Resolve capability and provider bindings against the registry
PipelineDefinitionModel.ResolutionResult resolution =
    model.resolveBindings(pipeline, providerRegistry);

if (!resolution.resolved()) {
    System.err.println("Unresolved bindings:");
    resolution.unresolvedBindings().forEach(b -> System.err.println("  - " + b));
    // Example output:
    // - Stage 'acquire-customers': unresolved provider reference 'jdbc-provider'
}

Step 5: Print (Serialize) the Pipeline

// Serialize back to YAML (round-trip safe)
String outputYaml = model.print(pipeline);
System.out.println(outputYaml);

The printed output is semantically equivalent to the original — parse(print(pipeline)) always yields the same Pipeline.

What's Not Allowed

PatternOps enforces technology-free pipeline definitions. These configs will be rejected:

# ❌ Spark-specific constructs
config:
  sparkConf:
    spark.sql.shuffle.partitions: 200
  numExecutors: 10
  driverMemory: 4g

# ❌ Airflow-specific constructs
config:
  dag_id: my_dag
  task_id: my_task
  operator: BashOperator

# ❌ AWS-specific constructs
config:
  s3Bucket: my-bucket
  lambdaArn: arn:aws:lambda:...
  glueJobName: my-job

# ❌ Databricks-specific constructs
config:
  clusterSpec: { size: large }
  notebookPath: /Workspace/notebook
  dbfsPath: dbfs:/data/output

All technology-specific behaviour belongs in Providers, not in pipeline definitions.

Next Steps