Starlake Orchestration

The core Python module for creating, scheduling, and managing data pipelines across multiple orchestration platforms.

What is Starlake?

Starlake is a configuration-driven platform designed to simplify Extract, Load, and Transform (ELT) operations while supporting declarative orchestration of data pipelines. By minimizing coding requirements, it empowers users to create robust data workflows with YAML-based configurations.

Typical Use Case

1. Extract: Gather data from sources such as Fixed Position files, DSV (Delimiter-Separated Values), JSON, or XML formats.
2. Define or infer structure: Use YAML to describe or infer the schema for each data source.
3. Load: Configure and execute the loading process to your data warehouse or other sinks.
4. Transform: Build aggregates and join datasets using SQL, Jinja, and YAML configurations.
5. Output: Observe your data becoming available as structured tables in your data warehouse.

Flexibility Across Workflows

Starlake supports any or all steps in your data pipeline, allowing for seamless integration into existing workflows:

• Extract: Export selective data from SQL databases into CSV files.
• Preload: Evaluate whether the loading process should proceed, based on a configurable preload strategy.
• Load: Ingest FIXED-WIDTH, CSV, JSON, or XML files, converting them into strongly-typed records stored as Parquet files, data warehouse tables (e.g., Google BigQuery), or other configured sinks.
• Transform: Join loaded datasets and save them as Parquet files, data warehouse tables, or Elasticsearch indices.

What is Starlake Orchestration?

Starlake Orchestration is a Python-based API for creating, scheduling, and managing data pipelines. It abstracts the complexities of various orchestration platforms, offering a unified interface for pipeline orchestration.

It is recommended to use it in combination with starlake dag generation, but it can also be used directly in your DAGs.

Key Features

1. Multi-Orchestrator Support

Starlake Orchestration integrates seamlessly with multiple orchestration frameworks, letting you select the best fit for your requirements.

2. Write Once, Deploy Anywhere

Design your pipelines once and execute them seamlessly across diverse orchestrators and environments without rewriting code. Starlake ensures consistent pipeline definitions, whether you are using Airflow, Dagster, or Snowflake on Google Cloud Platform (GCP), Amazon Web Services (AWS), or on-premises.

Run Starlake jobs effortlessly using GCP Cloud Run, GCP Dataproc, AWS Fargate, or simple shell scripts.

This flexibility empowers teams to:

• Transition seamlessly between execution environments.
• Integrate with cloud-native or on-premises orchestration tools.
• Simplify deployments without compromising functionality or performance.

3. Data Freshness and Scheduling

Starlake Orchestration supports flexible scheduling mechanisms, ensuring your data pipelines deliver up-to-date results:

• Cron-based Scheduling: Automate periodic pipeline runs (e.g., "Run at 2 AM daily").
• Event-Driven Orchestration: Dynamically trigger pipelines using dataset-aware DAGs, ensuring dependencies and lineage are respected.

By leveraging data lineage and dependencies, Starlake Orchestration aligns schedules automatically, ensuring the freshness of interconnected datasets.

4. Simplified Management

With automated schedule alignment and dependency management, Starlake Orchestration eliminates manual adjustments and simplifies pipeline workflows while maintaining reliability.

Prerequisites

• Starlake CLI 1.5.7+
• Python 3.8+

Installation

pip install starlake-orchestration

Package Structure

The core module provides the following packages under ai.starlake:

Main Components

1. IStarlakeJob[T, E]

ai.starlake.job.IStarlakeJob serves as the generic factory interface for creating orchestration tasks. These tasks execute the appropriate Starlake CLI commands, allowing seamless integration with orchestration platforms.

Classification Methods

@classmethod
def sl_orchestrator(cls) -> Union[StarlakeOrchestrator, str, None]:
    """Returns the orchestrator type for this job implementation."""

@classmethod
def sl_execution_environment(cls) -> Union[StarlakeExecutionEnvironment, str, None]:
    """Returns the execution environment type for this job implementation."""

Core Abstract Method

@abstractmethod
def sl_job(
    self,
    task_id: str,
    arguments: list,
    spark_config: Optional[StarlakeSparkConfig] = None,
    dataset: Optional[Union[StarlakeDataset, str]] = None,
    task_type: Optional[TaskType] = None,
    **kwargs
) -> T:
    """Create an orchestrator-specific task that runs a Starlake CLI command."""

Factory Methods for Core Starlake Commands

Pre-load

Generates the task that will run the starlake preload command.

def sl_pre_load(
    self,
    domain: str,
    tables: set = set(),
    pre_load_strategy: Union[StarlakePreLoadStrategy, str, None] = None,
    sensor: Optional[bool] = None,
    **kwargs
) -> Optional[T]:

StarlakePreLoadStrategy

ai.starlake.job.StarlakePreLoadStrategy is an enumeration defining preload strategies for conditional domain loading.

1. NONE -- No condition applied; preload tasks are skipped.

   

2. IMPORTED -- Load only if files exist in the landing area (SL_ROOT/datasets/importing/{domain}).

   

3. PENDING -- Load only if files exist in the pending datasets area (SL_ROOT/datasets/pending/{domain}).

   

4. ACK -- Load only if an acknowledgment file exists at the configured path (global_ack_file_path).

   

IMPORTED chain: sl_pre_load >> skip_or_start >> sl_import >> sl_load

Pre-load not-ready sentinel

Starlake CLI 1.5.15+ supports preload --notReadySentinel <uri>: on a "not ready" decision (IMPORTED/PENDING empty, ACK file missing) the CLI touches a zero-byte marker at exactly the given URI and exits 0; on "ready" no marker is written; on a genuine crash the process still exits non-zero and never writes the marker. This turns the lossy exit-code channel into a deterministic verdict.

The framework consumes this contract through the opt-in pre_load_not_ready_sentinel_path option (a parent prefix — absent or blank keeps everything byte-identical to today). When set, sl_pre_load appends --notReadySentinel <prefix>/<domain>/<scope>.notready to the CLI arguments for all three strategies, where <scope> is the sanitized run scope (<dag_id>__<run_id> on Airflow, <job_name>__<run_id> on Dagster; whitelist [A-Za-z0-9_.+:=-], anything else mapped to _) substituted at RUN time — never via templating. The uniform verdict wherever the preload outcome is interpreted:

• exit 0 + marker absent → READY → proceed;
• exit 0 + marker present → consume (delete first), then signal NOT READY through the mechanism's existing primitive (skip via skip_or_start, poke-again in sensor/poke-loop/cloud waiting);
• non-zero / engine failure → REAL FAILURE → fail now (opt-in behavior improvement: a crashed CLI no longer reads as "nothing to load", and broken invocations no longer poke until timeout).

Scheme support is engine-gated at definition time: local absolute paths (or file://) on shell, gs:// on cloud_run/dataproc, s3:// on fargate — any mismatch raises a ValueError naming the engine. Default consumption handlers ship in core: local handlers in ai.starlake.sentinel, GCS/S3 handler factories in ai.starlake.gcp/ai.starlake.aws with lazy SDK imports — install them with the extras pip install starlake-orchestration[gcp] (google-cloud-storage) or starlake-orchestration[aws] (boto3). The Airflow cloud paths override the defaults with GCSHook/S3Hook so gcp_conn_id/aws_conn_id and impersonation_chain are honored.

Best-effort-write caveat (CLI design): a failed marker write is logged and swallowed CLI-side — "exit 0 + no marker" can, rarely, mean "not ready but the write failed". For IMPORTED/PENDING this yields a harmless no-op load (the pending area is empty by definition of not-ready). For ACK it is sharper: the ack file was missing but the pending area may hold real files, so a lost write can trigger a premature load of un-acked data — keep the sentinel prefix on reliable storage when using the ACK strategy.

Import

Generates the task for the import command.

def sl_import(
    self,
    task_id: str,
    domain: str,
    tables: set = set(),
    **kwargs
) -> T:

Load

Generates the task for the load command.

def sl_load(
    self,
    task_id: str,
    domain: str,
    table: str,
    spark_config: Optional[StarlakeSparkConfig] = None,
    dataset: Optional[Union[StarlakeDataset, str]] = None,
    **kwargs
) -> T:

Transform

Generates the task for the transform command.

def sl_transform(
    self,
    task_id: str,
    transform_name: str,
    transform_options: str = None,
    spark_config: Optional[StarlakeSparkConfig] = None,
    dataset: Optional[Union[StarlakeDataset, str]] = None,
    **kwargs
) -> T:

2. StarlakeDataset

ai.starlake.dataset.StarlakeDataset represents the metadata of a dataset produced by a task.

Starlake Orchestration collects all datasets produced by each task into a list of events to trigger per task. At runtime, the orchestrator triggers subsequent events only if their corresponding tasks succeed.

Key properties:

3. StarlakeOptions

ai.starlake.job.StarlakeOptions provides methods to manage and retrieve configuration variables. Variables are resolved in order: options dict -> default_value -> environment variable.

The following options are available for all concrete factory classes derived from IStarlakeJob:

4. Abstract Classes

AbstractDependency

Defines task dependencies, ensuring execution order in Directed Acyclic Graphs (DAGs). Operators >> and << allow intuitive chaining and automatically register dependencies in the current TaskGroupContext.

AbstractTask[T]

Wraps concrete orchestration tasks into a unified interface. Auto-registers with the current TaskGroupContext on creation.

AbstractTaskGroup[GT]

Groups related tasks into cohesive units. Supports nested grouping via context managers (with blocks).

AbstractPipeline[U, T, GT, E]

Defines an entire pipeline, combining tasks and task groups. It handles:

• Task management: Adding and managing orchestrator-specific tasks via @final methods (sl_load, sl_transform, sl_import, sl_pre_load).
• Dependency management: Ensuring the correct execution order.
• Lifecycle: start_task(), end_task(), pre_tasks(), post_tasks().
• Execution: Only run() is abstract -- all other lifecycle methods (deploy(), delete(), dry_run(), backfill()) are concrete.

Pipelines are constructed with either:

• StarlakeSchedule -- time-driven: cron expression + list of domains/tables.
• StarlakeDependencies -- data-driven: parsed from starlake dag-generate JSON, with dependency graphs and dataset events.

AbstractOrchestration[U, T, GT, E]

The central abstraction for creating pipelines, tasks, and task groups. Orchestrator-specific implementations extend this class.

Key methods:

• sl_orchestrator() -- Returns the orchestrator type.
• sl_create_pipeline(schedule, dependencies, ...) -- Creates a pipeline instance.
• sl_create_task_group(group_id, pipeline, ...) -- Creates task groups for organizing related tasks.

5. TaskGroupContext

A context manager responsible for:

• Tracking the current task group or pipeline context via a static _context_stack.
• Automatically adding tasks to the active group.
• Managing dependencies within the group (upstream/downstream relationships, roots, leaves).

6. StarlakeDependencies

Parses JSON dependency graphs (produced by starlake dag-generate) into traversable StarlakeDependency trees. Key capabilities:

• graphs() -- Returns traversable TreeNodeMixin trees.
• get_schedule() -- Computes scheduling from dependencies.
• retrieve_datasets() -- Extracts datasets for event-driven triggering.

7. Factories

OrchestrationFactory

Handles the dynamic registration and instantiation of concrete orchestration classes.

class OrchestrationFactory:
    @classmethod
    def register_orchestrations_from_package(cls, package_name: str = "ai.starlake") -> None:
        """Dynamically load all AbstractOrchestration subclasses from the given package."""

    @classmethod
    def register_orchestration(cls, orchestration_class: Type[AbstractOrchestration]) -> None:
        """Manually register an orchestration class."""

    @classmethod
    def create_orchestration(cls, job: IStarlakeJob, **kwargs) -> AbstractOrchestration:
        """Create the correct AbstractOrchestration instance based on job.sl_orchestrator()."""

StarlakeJobFactory

Handles the dynamic registration and instantiation of concrete IStarlakeJob classes.

class StarlakeJobFactory:
    @classmethod
    def register_jobs_from_package(cls, package_name: str = "ai.starlake") -> None:
        """Dynamically load all IStarlakeJob subclasses from the given package."""

    @classmethod
    def register_job(cls, job_class: Type[IStarlakeJob]) -> None:
        """Manually register a job class by (orchestrator, execution_environment)."""

    @classmethod
    def create_job(
        cls,
        filename: str,
        module_name: str,
        orchestrator: Union[StarlakeOrchestrator, str],
        execution_environment: Union[StarlakeExecutionEnvironment, str],
        options: dict,
        **kwargs
    ) -> IStarlakeJob:
        """Create the correct IStarlakeJob instance."""

SessionFactory

Creates database sessions for SQL-based orchestration and testing.

class SessionFactory:
    @classmethod
    def session(
        cls,
        provider: SessionProvider = SessionProvider.DUCKDB,
        database: Optional[str] = None,
        schema: Optional[str] = None,
        user: Optional[str] = None,
        password: Optional[str] = None,
        host: Optional[str] = None,
        port: Optional[int] = None,
        **kwargs
    ) -> Session:
        """Create a new session based on the provider."""

Supported providers via SessionProvider: DUCKDB, POSTGRES, MYSQL, REDSHIFT, SNOWFLAKE, BIGQUERY.

Enums Reference

CLI

The module includes a command-line interface for executing pipelines directly:

python -m ai.starlake.orchestration <action> --file <path> [--options <options>]

Supported actions: run, dry-run, deploy, delete, backfill.

Example:

python -m ai.starlake.orchestration run --file /path/to/my_dag.py
python -m ai.starlake.orchestration dry-run --file /path/to/dags/ --options '{"key": "value"}'
python -m ai.starlake.orchestration backfill --file /path/to/my_dag.py --options 'start_date=2024-01-01,end_date=2024-01-31'

How to Extend Starlake Orchestration

1. Define a Starlake Job

Implement the IStarlakeJob interface to create a concrete factory class responsible for defining orchestrator-specific tasks.

from ai.starlake.job import IStarlakeJob, StarlakeOrchestrator, StarlakeExecutionEnvironment

class MyStarlakeJob(IStarlakeJob):
    @classmethod
    def sl_orchestrator(cls) -> str:
        return "my_orchestrator"

    @classmethod
    def sl_execution_environment(cls) -> str:
        return "shell"

    def sl_job(self, task_id, arguments, spark_config=None, dataset=None, task_type=None, **kwargs):
        return MyOrchestratorTask(task_id=task_id, command_arguments=arguments, **kwargs)

    def dummy_op(self, task_id, events=None, task_type=None, **kwargs):
        return MyDummyTask(task_id=task_id, **kwargs)

    def skip_or_start_op(self, task_id, upstream_task, **kwargs):
        return None  # or a conditional task

    def to_event(self, dataset, source=None, **kwargs):
        return MyEvent(dataset=dataset, source=source)

2. Implement the Orchestration API

Extend AbstractOrchestration to integrate the new orchestrator's API.

from ai.starlake.orchestration import AbstractPipeline, AbstractTaskGroup, AbstractOrchestration

class MyPipeline(AbstractPipeline):
    def run(self, **kwargs):
        # Execute the pipeline using the orchestrator's API
        ...

class MyTaskGroup(AbstractTaskGroup):
    ...

class MyOrchestration(AbstractOrchestration):
    @classmethod
    def sl_orchestrator(cls) -> str:
        return "my_orchestrator"

    def sl_create_pipeline(self, schedule=None, dependencies=None, **kwargs):
        return MyPipeline(self.job, schedule=schedule, dependencies=dependencies, orchestration=self, **kwargs)

    def sl_create_task_group(self, group_id, pipeline, **kwargs):
        return MyTaskGroup(name=group_id, pipeline=pipeline, **kwargs)

3. Register (Optional)

Registration happens automatically via importlib package discovery when using the factories. For explicit registration:

from ai.starlake.orchestration import OrchestrationFactory
from ai.starlake.job import StarlakeJobFactory

OrchestrationFactory.register_orchestration(MyOrchestration)
StarlakeJobFactory.register_job(MyStarlakeJob)

4. Create and Run a Pipeline

from ai.starlake.job import StarlakeJobFactory, StarlakeExecutionEnvironment
from ai.starlake.orchestration import StarlakeSchedule, StarlakeDomain, StarlakeTable, OrchestrationFactory
from ai.starlake.common import sanitize_id

import os

schedule = StarlakeSchedule(
    name='daily',
    cron='0 0 * * *',
    domains=[
        StarlakeDomain(
            name='starbake',
            final_name='starbake',
            tables=[
                StarlakeTable(name='Customers', final_name='Customers'),
                StarlakeTable(name='Products', final_name='Products'),
            ]
        )
    ]
)

sl_job = StarlakeJobFactory.create_job(
    filename=os.path.basename(__file__),
    module_name=f"{__name__}",
    orchestrator="my_orchestrator",
    execution_environment=StarlakeExecutionEnvironment.SHELL,
    options={}
)

with OrchestrationFactory.create_orchestration(job=sl_job) as orchestration:

    with orchestration.sl_create_pipeline(schedule=schedule) as pipeline:

        start = pipeline.start_task()

        def generate_load_domain(domain: StarlakeDomain):
            with orchestration.sl_create_task_group(group_id=sanitize_id(domain.name), pipeline=pipeline) as ld:
                with orchestration.sl_create_task_group(group_id=sanitize_id(f'load_{domain.name}'), pipeline=pipeline) as load_tables:
                    for table in domain.tables:
                        pipeline.sl_load(
                            task_id=sanitize_id(f'load_{domain.name}_{table.name}'),
                            domain=domain.name,
                            table=table.name,
                        )
                return load_tables
            return ld

        load_domains = [generate_load_domain(domain) for domain in schedule.domains]

        start >> load_domains

        end = pipeline.end_task()
        end << load_domains

Integration Modules

The core starlake-orchestration module is extended by orchestrator-specific integration modules:

Each integration module implements IStarlakeJob, AbstractOrchestration, AbstractPipeline, and AbstractTaskGroup for its target platform. See the individual module READMEs for platform-specific documentation and examples.

Architecture

For detailed architectural documentation including the inheritance hierarchy, design patterns (double factory, context stack, @final invariants), and class-by-class reference, see ARCHITECTURE.md.