lakeflow-jobs-meta 0.2.0

Metadata-driven framework for orchestrating Databricks Lakeflow Jobs

Databricks Lakeflow Jobs Meta

A metadata-driven framework for orchestrating Databricks Lakeflow Jobs. Package as a library and run as a single task in a Lakeflow Jobs to continuously monitor for metadata changes and automatically update jobs.

Features

• ✅ Dynamic Job Generation: Automatically creates/updates Databricks jobs from metadata
• ✅ Variable Substitution: Use ${var.name} syntax for dynamic YAML templates with per-job error isolation
• ✅ Resource ID Tracking: Stable job tracking independent of job name changes
• ✅ Continuous Monitoring: Automatically detects metadata changes and updates jobs
• ✅ Multiple Task Types: Support for Notebook, SQL Query, SQL File, Python Wheel, Spark JAR, Pipeline, and dbt tasks
• ✅ Advanced Task Features: Support for run_if conditions, job clusters, environments, and notifications
• ✅ Trigger and Schedule Support: File arrival, table update, periodic triggers, and cron schedules with automatic enum conversion
• ✅ Delta Table as Source of Truth: Manage metadata directly in Delta tables
• ✅ YAML Support: YAML file ingestion for bulk updates
• ✅ Dependency Management: Handles execution order and task dependencies
• ✅ Job Lifecycle: Tracks and manages job IDs in Delta tables
• ✅ Robust Error Handling: Job-level error isolation ensures invalid jobs don't block valid ones

Architecture

┌─────────────────────────────────────────┐
│ Monitoring Job                          │
│  - Watches Delta Table                  │
│  - Watches Unity Catalog Volume (YAML)  │
│  - Auto-updates Jobs on Changes         │
└─────────────────────────────────────────┘
           │                    │
           ▼                    ▼
    Delta Control Table    YAML Files (UC Volume)
           │                    │
           └──────────┬─────────┘
                      ▼
              Job Generator
                      ▼
            Databricks Jobs

Package Structure

.
├── lakeflow_jobs_meta/         # Main package
│   ├── __init__.py
│   ├── main.py                 # Entry point for monitoring task
│   ├── constants.py
│   ├── utils.py
│   ├── task_builders.py
│   ├── orchestrator.py
│   ├── metadata_manager.py
│   └── monitor.py
├── examples/                     # Example files
│   ├── orchestrator_example.ipynb  # Orchestrator example notebook
│   ├── notebook_task/          # Example notebook tasks
│   │   └── sample_ingestion_notebook.ipynb # Example ingestion notebook task
│   ├── sql_file_task/          # SQL file task examples
│   │   ├── 01_create_sample_data.sql
│   │   ├── 02_daily_aggregations.sql
│   │   ├── 03_bronze_to_silver_transformation.sql
│   │   ├── 04_data_freshness_check.sql
│   │   └── 05_incremental_load.sql
│   └── metadata_examples.yaml   # Example metadata configurations
├── docs/                        # Documentation
│   ├── PACKAGING_AND_DEPLOYMENT.md
│   └── METADATA_MANAGEMENT.md
├── tests/                       # Test suite
│   ├── test_utils.py
│   ├── test_task_builders.py
│   ├── test_metadata_manager.py
│   ├── test_orchestrator.py
│   ├── test_monitor.py
│   └── test_constants.py
├── setup.py                     # Package setup
├── pyproject.toml              # Modern Python packaging

Quick Start

Installation

# Install from PyPI
pip install lakeflow-jobs-meta

# Or install from source (development)
pip install -e .

# Or install from wheel
pip install dist/lakeflow_jobs_meta-0.2.0-py3-none-any.whl

Quick Example

import lakeflow_jobs_meta as jm

# Example 1: Load metadata from YAML with variables
vars = {
    'env': 'prod',
    'warehouse_id': 'abc123',
    'catalog': 'bronze'
}

jobs = jm.create_or_update_jobs(
    yaml_path="/Workspace/path/to/metadata.yaml",
    control_table="catalog.schema.etl_control",
    var=vars
)

# Example 2: Load from folder (all YAML files)
jobs = jm.create_or_update_jobs(
    yaml_path="/Workspace/path/to/metadata/",
    control_table="catalog.schema.etl_control"
)

# Example 3: Create/update all jobs in control table
jobs = jm.create_or_update_jobs(
    control_table="catalog.schema.etl_control"
)

YAML Format

Jobs are defined as dictionary entries with resource IDs as keys:

jobs:
  # resource_id is the YAML dict key
  customer_etl_pipeline:
    name: "Customer ETL - Production"  # Optional: Databricks job name (defaults to resource_id)
    description: "Daily customer data pipeline"
    schedule:
      quartz_cron_expression: "0 0 2 * * ?"
      timezone_id: "UTC"
    tasks:
      - task_key: "extract_customers"
        task_type: "sql_query"
        warehouse_id: "abc123"
        sql_query: "SELECT * FROM bronze.raw.customers"

Variable Substitution

Use ${var.name} syntax for dynamic values anywhere in your YAML:

jobs:
  etl_${var.env}_${var.source}:
    name: "ETL Pipeline - ${var.source} (${var.env})"
    tasks:
      - task_key: "extract_${var.source}"
        task_type: "sql_query"
        warehouse_id: "${var.warehouse_id}"
        sql_query: "SELECT * FROM ${var.catalog}.${var.schema}.${var.source}"

Pass variables when loading:

vars = {
    'env': 'prod',
    'source': 'customers',
    'warehouse_id': 'abc123',
    'catalog': 'bronze',
    'schema': 'raw'
}

jobs = jm.create_or_update_jobs(
    yaml_path="/path/to/template.yaml",
    control_table="catalog.schema.control",
    var=vars
)

Important Notes:

• Variable substitution happens per-job after YAML parsing
• Comments containing ${var.xxx} won't cause errors
• Undefined variables will skip that job with a warning and continue processing other jobs
• Use ${var.name} syntax (curly braces), not $(var.name) (parentheses)

### Usage as a Lakeflow Jobs Task (Recommended)

1. **Use the orchestrator example** from `examples/orchestrator_example.ipynb` to create/update jobs on-demand, OR create a continuous monitoring job as shown below

2. **Configure Parameters** via Databricks widgets or base_parameters:
   ```python
   {
       "control_table": "your_catalog.schema.etl_control",  # Required
       "volume_path": "/Volumes/catalog/schema/metadata",  # Optional
       "check_interval_seconds": "60",  # Optional, default: 60
       "max_iterations": ""  # Optional, empty = infinite
   }

3. Run the Job - It will continuously monitor for changes and auto-update jobs

Option 1: YAML File Ingestion (Recommended)

The yaml_path parameter accepts three types of paths:

1. Single YAML File

import lakeflow_jobs_meta as jm

# Load and process jobs from a single YAML file
jobs = jm.create_or_update_jobs(
    yaml_path="/Workspace/path/to/metadata.yaml",
    control_table="your_catalog.schema.etl_control"
)

2. Folder Path (All YAML Files)

import lakeflow_jobs_meta as jm

# Load and process all YAML files in folder (recursive)
jobs = jm.create_or_update_jobs(
    yaml_path="/Workspace/path/to/metadata/",
    control_table="your_catalog.schema.etl_control"
)

3. Unity Catalog Volume with File Arrival Trigger (Recommended for Production)

For production environments, use Databricks file arrival triggers to automatically process YAML files when they're uploaded to a Unity Catalog volume:

1. Configure File Arrival Trigger: Set up a file arrival trigger on your job to monitor the Unity Catalog volume. See Databricks File Arrival Triggers.

2. Upload YAML Files: When YAML files are uploaded, the job automatically triggers.

3. Process YAML Files: The job loads and processes all YAML files from the volume:

import lakeflow_jobs_meta as jm

# Automatically runs when file arrival trigger fires
jobs = jm.create_or_update_jobs(
    yaml_path="/Volumes/catalog/schema/metadata_volume",
    control_table="your_catalog.schema.etl_control"
)

Advanced Usage: Load Separately

You can also load YAML separately from orchestration:

import lakeflow_jobs_meta as jm

# Load from file
num_tasks, job_names = jm.load_yaml("/Workspace/path/to/metadata.yaml")

# Load from folder
num_tasks, job_names = jm.load_from_folder("/Workspace/path/to/metadata/")

# Load from volume
num_tasks, job_names = jm.sync_from_volume("/Volumes/catalog/schema/volume")

# Then process all jobs in control table
jobs = jm.create_or_update_jobs(control_table="your_catalog.schema.etl_control")

Option 2: Direct Delta Table Updates

For advanced use cases, you can update metadata directly in the Delta table:

-- Insert new task
INSERT INTO your_catalog.schema.etl_control VALUES (
  'my_pipeline',           -- job_name
  'sql_task_1',            -- task_key
  '[]',                     -- depends_on (JSON array of task_keys, empty for no dependencies)
  'sql_query',             -- task_type
  '{"timeout_seconds": 7200, "tags": {"department": "engineering"}}', -- job_config (JSON string)
  '{"warehouse_id": "abc123", "sql_query": "SELECT * FROM bronze.customers", "parameters": {"catalog": "my_cat"}}', -- task_config (JSON string)
  false                    -- disabled
);

The monitoring job will automatically detect the change and update the job.

Option 3: On-Demand Orchestration

Run the orchestrator manually (for development/testing):

import lakeflow_jobs_meta as jm

# Process all jobs in control table
jobs = jm.create_or_update_jobs(
    control_table="your_catalog.schema.etl_control"
)

# Or load from YAML and process those jobs
jobs = jm.create_or_update_jobs(
    yaml_path="/Workspace/path/to/metadata.yaml",
    control_table="your_catalog.schema.etl_control"
)

# With custom configuration
jobs = jm.create_or_update_jobs(
    yaml_path="/Workspace/path/to/metadata/",
    control_table="your_catalog.schema.control_table",
    jobs_table="your_catalog.schema.custom_jobs_table",
    default_warehouse_id="your-warehouse-id",
    default_pause_status=False
)

Note: When yaml_path is provided, only jobs from that path are processed. When not provided, all jobs in the control table are processed.

Pause Status Management

The default_pause_status parameter controls the initial behavior of newly created jobs:

default_pause_status=False (default behavior):

• Manual jobs (no schedule/trigger/continuous): Auto-run immediately after creation
• Jobs with schedule/trigger/continuous: Created active (UNPAUSED), run according to schedule

default_pause_status=True:

• Manual jobs: Do NOT auto-run after creation
• Jobs with schedule/trigger/continuous: Created paused (PAUSED), won't run until unpaused

Explicit pause_status in YAML always overrides the default:

The pause_status is set within the continuous, schedule, or trigger configuration:

jobs:
  scheduled_job:
    name: "Scheduled Job"
    continuous:
      pause_status: UNPAUSED  # Explicit - overrides default_pause_status
      task_retry_mode: ON_FAILURE
    tasks:
      - task_key: "my_task"
        # ...
  
  cron_job:
    name: "Cron Job"
    schedule:
      quartz_cron_expression: "0 0 2 * * ?"
      timezone_id: "America/Los_Angeles"
      pause_status: PAUSED  # Explicit - job won't run until unpaused
    tasks:
      - task_key: "my_task"
        # ...

Behavior:

• If pause_status is explicitly set: That value is used (for both create and update)
• If pause_status is NOT set:
  • For NEW jobs: default_pause_status is applied
  • For UPDATES: Existing pause status is unchanged

For Job Updates:

• default_pause_status has NO effect on job updates
• Jobs are never auto-run when updating existing jobs
• Pause status only changes if explicitly set in YAML metadata
• This prevents accidentally pausing or running production jobs during updates

UI Lock Protection:

By default, all jobs created by this framework are set with edit_mode: UI_LOCKED, which means:

• ✅ Jobs cannot be edited in the Databricks UI (default)
• ✅ Prevents accidental manual modifications
• ✅ Enforces metadata as the single source of truth
• ✅ Users can still view job details and run history
• ✅ Changes must be made in YAML metadata

Override Edit Mode (Optional):

If you need to allow UI editing for specific jobs, you can set edit_mode: EDITABLE in your YAML:

jobs:
  experimental_job:
    name: "Experimental Job"
    edit_mode: EDITABLE  # Allow UI editing for this job
    tasks:
      - task_key: "my_task"
        # ...

Note: It's recommended to keep edit_mode: UI_LOCKED (default) for production jobs to maintain consistency.

Best Practice:

• Always define all job settings in YAML metadata
• All changes must go through the metadata workflow
• Use version control (Git) to track metadata changes
• Use metadata as the single source of truth

Testing

Running Tests

# Install dev dependencies
pip install -e ".[dev]"

# Run all tests
pytest

# Run with coverage
pytest --cov=lakeflow_jobs_meta --cov-report=html

# Run specific test file
pytest tests/test_utils.py

# Run specific test
pytest tests/test_utils.py::TestSanitizeTaskKey::test_basic_sanitization

Test Structure

Tests are organized in the tests/ directory:

• test_utils.py - Utility function tests
• test_task_builders.py - Task creation tests
• test_metadata_manager.py - Metadata management tests
• test_orchestrator.py - Orchestration tests
• test_monitor.py - Monitoring tests
• test_constants.py - Constants validation

All tests use pytest with mocking for external dependencies (Databricks SDK, Spark, dbutils).

Metadata Schema

YAML Structure

Jobs are defined as a dictionary where each key is a unique resource_id:

jobs:
  <resource_id>:        # Unique stable identifier (e.g., customer_etl_pipeline)
    name: <string>      # Optional: Databricks job display name (defaults to resource_id)
    description: <string>
    # ... other job-level configuration
    tasks:
      - task_key: <string>
        task_type: <string>
        # ... task configuration

Job-Level Configuration

Each job in your YAML is defined with the following structure:

Required Fields

Parameter	Type	Description	Example
resource_id	String (YAML key)	Unique identifier for the job	customer_etl_pipeline
tasks	List	List of task definitions (see Task Configuration below)	See Task Configuration

Optional Fields

Parameter	Type	Default	Description	Possible Values
name	String	resource_id	Databricks job display name	Any string
description	String	None	Human-readable description of the job	Any string
timeout_seconds	Integer	7200	Maximum time the entire job can run (seconds)	Any positive integer
max_concurrent_runs	Integer	1	Maximum number of concurrent runs allowed	Any positive integer
edit_mode	String	"UI_LOCKED"	Whether job can be edited in Databricks UI	"UI_LOCKED", "EDITABLE"
parameters	Object	None	Job-level parameters (automatically passed to tasks)	Key-value pairs
tags	Object	None	Job tags for organization and filtering	Key-value pairs
queue	Object	None	Job queue settings	{enabled: true/false}
continuous	Object	None	Continuous job settings (always-on)	See Continuous Settings below
trigger	Object	None	Job trigger settings (file arrival, table update, or periodic)	See Trigger Settings below
schedule	Object	None	Scheduled job settings (cron-based)	See Schedule Settings below
job_clusters	List	None	Shared cluster definitions for tasks	See Job Clusters below
environments	List	None	Python/Scala environment definitions	See Environments below
notification_settings	Object	None	Job-level notification settings	See Notification Settings below

Continuous Settings

Used for always-on streaming jobs:

Parameter	Type	Default	Description	Possible Values
pause_status	String	"UNPAUSED"	Initial pause state	"UNPAUSED", "PAUSED"
task_retry_mode	String	"ON_FAILURE"	When to retry failed tasks	"ON_FAILURE", "NEVER"

Example:

continuous:
  pause_status: UNPAUSED
  task_retry_mode: ON_FAILURE

Schedule Settings

Used for cron-scheduled jobs:

Parameter	Type	Required	Description	Example
quartz_cron_expression	String	Yes	Quartz cron expression	"0 0 2 * * ?" (daily at 2am)
timezone_id	String	Yes	Timezone for the schedule	"UTC", "America/Los_Angeles"
pause_status	String	No	Initial pause state	"UNPAUSED", "PAUSED"

Example:

schedule:
  quartz_cron_expression: "0 0 2 * * ?"
  timezone_id: "America/Los_Angeles"
  pause_status: UNPAUSED

Trigger Settings

Used for event-driven jobs (file arrival, table update, or periodic):

Trigger Type	Parameters	Description	Example
File Arrival	file_arrival.url	Trigger when files arrive at path	url: /Volumes/catalog/schema/folder/
Table Update	table_update.table_names	Trigger when tables are updated	table_names: ["catalog.schema.table"]
Periodic	periodic.interval periodic.unit	Trigger at regular intervals	interval: 1 unit: HOURS
All Types	pause_status	Initial pause state	UNPAUSED or PAUSED

Example:

trigger:
  pause_status: UNPAUSED
  file_arrival:
    url: /Volumes/catalog/schema/metadata/

Job Clusters

Define shared cluster configurations that tasks can reference:

Parameter	Type	Description	Example
job_cluster_key	String	Unique identifier for this cluster	"Job_cluster"
new_cluster	Object	Full cluster specification	See Databricks cluster API docs

Example:

job_clusters:
  - job_cluster_key: "Job_cluster"
    new_cluster:
      spark_version: "16.4.x-scala2.12"
      node_type_id: "rd-fleet.xlarge"
      num_workers: 8
      aws_attributes:
        availability: SPOT_WITH_FALLBACK
      data_security_mode: SINGLE_USER
      runtime_engine: PHOTON

Environments

Define Python/Scala environments that tasks can reference:

Parameter	Type	Description	Example
environment_key	String	Unique identifier for this environment	"default_python"
spec.dependencies	List	Python dependencies (pip packages)	["dbt-databricks>=1.0.0"]
spec.java_dependencies	List	Java/Scala dependencies (JARs)	["/Workspace/.../file.jar"]
spec.environment_version	String	Databricks environment version	"4", "4-scala-preview"

Example:

environments:
  - environment_key: "default_python"
    spec:
      dependencies:
        - "dbt-databricks>=1.0.0"
        - "pandas>=1.5.0"
      environment_version: "4"

Notification Settings

Configure job-level email notifications:

Parameter	Type	Description	Example
email_notifications.on_start	List	Email addresses to notify on job start	["[e-mail]"]
email_notifications.on_success	List	Email addresses to notify on success	["[e-mail]"]
email_notifications.on_failure	List	Email addresses to notify on failure	["[e-mail]"]
email_notifications.on_duration_warning_threshold_exceeded	List	Email addresses for duration warnings	["[e-mail]"]
no_alert_for_skipped_runs	Boolean	Skip alerts for skipped runs	true, false
no_alert_for_canceled_runs	Boolean	Skip alerts for canceled runs	true, false
alert_on_last_attempt	Boolean	Alert only on last retry attempt	true, false

Example:

notification_settings:
  email_notifications:
    on_failure: ["[e-mail]"]
    on_duration_warning_threshold_exceeded: ["[e-mail]"]
  alert_on_last_attempt: true

Task Configuration

Each task within a job has the following structure:

Required Fields

Parameter	Type	Description	Example
task_key	String	Unique identifier for this task within the job	"ingest_data"
task_type	String	Type of task to execute	See Task Types section below

Common Optional Fields

These fields are available for all task types:

Parameter	Type	Default	Description	Possible Values
depends_on	List	[]	List of task_key values this task depends on	["task_a", "task_b"]
disabled	Boolean	false	Whether this task is disabled	true, false
timeout_seconds	Integer	3600	Maximum time this task can run (seconds)	Any positive integer
run_if	String	"ALL_SUCCESS"	Condition for when this task should run	See Run Conditions below
job_cluster_key	String	None	Reference to job-level cluster definition	Must match a job_clusters key
existing_cluster_id	String	None	Reference to existing all-purpose cluster	Databricks cluster ID
environment_key	String	None	Reference to job-level environment	Must match an environments key
notification_settings	Object	None	Task-level notification settings	Same structure as job-level
parameters	Object	None	Task-specific parameters	Key-value pairs or list (depends on task type)

Note: Use either job_cluster_key OR existing_cluster_id, not both.

Run Conditions

The run_if parameter controls when a task executes based on dependency outcomes:

Value	Description
ALL_SUCCESS	Run only if all dependencies succeed (default)
AT_LEAST_ONE_SUCCESS	Run if at least one dependency succeeds
NONE_FAILED	Run if no dependencies fail (success or skipped)
ALL_DONE	Run when all dependencies complete (regardless of status)
AT_LEAST_ONE_FAILED	Run if at least one dependency fails
ALL_FAILED	Run only if all dependencies fail

Complete Example

jobs:
  customer_etl_pipeline:
    name: "Customer ETL Pipeline - Production"
    description: "Daily ETL pipeline for customer data"
    timeout_seconds: 7200
    max_concurrent_runs: 1
    edit_mode: UI_LOCKED
    parameters:
      default_catalog: "production"
    tags:
      department: "engineering"
      project: "customer_analytics"
    schedule:
      quartz_cron_expression: "0 0 2 * * ?"
      timezone_id: "UTC"
      pause_status: UNPAUSED
    job_clusters:
      - job_cluster_key: "etl_cluster"
        new_cluster:
          spark_version: "16.4.x-scala2.12"
          node_type_id: "rd-fleet.xlarge"
          num_workers: 4
    tasks:
      - task_key: "extract_data"
        task_type: "notebook"
        file_path: "/Workspace/etl/extract_customers"
        job_cluster_key: "etl_cluster"
        parameters:
          source_table: "raw.customers"
      
      - task_key: "transform_data"
        task_type: "sql_query"
        depends_on: ["extract_data"]
        warehouse_id: "abc123"
        sql_query: "SELECT * FROM bronze.customers WHERE updated_date = CURRENT_DATE()"
      
      - task_key: "load_data"
        task_type: "notebook"
        depends_on: ["transform_data"]
        file_path: "/Workspace/etl/load_customers"
        run_if: "ALL_SUCCESS"
        notification_settings:
          email_notifications:
            on_failure: ["[e-mail]"]

Control Table Schema

The control table stores job and task metadata with the following schema:

CREATE TABLE control_table (
    resource_id STRING,           -- Stable tracking identifier (YAML dict key)
    job_name STRING,              -- Databricks job display name (from 'name' field or defaults to resource_id)
    task_key STRING,              -- Unique task identifier within the job
    depends_on STRING,            -- JSON array of task_key strings this task depends on
    task_type STRING,             -- Task type: notebook, sql_query, sql_file, python_wheel, spark_jar, pipeline, or dbt
    job_config STRING,            -- JSON string with job-level settings (tags, parameters, timeout_seconds, etc.)
    task_config STRING,           -- JSON string with task-specific config including parameters
    disabled BOOLEAN DEFAULT false, -- Whether task is disabled
    created_by STRING,            -- Username who created the record
    created_timestamp TIMESTAMP DEFAULT current_timestamp(),
    updated_by STRING,            -- Username who last updated the record
    updated_timestamp TIMESTAMP DEFAULT current_timestamp()
)

Key Concepts:

• resource_id: The stable tracking identifier defined as the YAML dictionary key. This never changes and is used to track job history in the control table.
• job_name: The display name shown in Databricks. Defaults to resource_id if the name field is not specified. Can be changed via the name field without losing job history.

Jobs Table Schema

The jobs tracking table maintains the mapping between resource IDs and Databricks job IDs:

CREATE TABLE jobs_table (
    resource_id STRING,           -- Stable tracking identifier (same as control table)
    job_id BIGINT,                -- Databricks job ID
    job_name STRING,              -- Databricks job display name (for reference)
    created_by STRING,            -- Username who created the record
    created_timestamp TIMESTAMP DEFAULT current_timestamp(),
    updated_by STRING,            -- Username who last updated the record
    updated_timestamp TIMESTAMP DEFAULT current_timestamp()
)

Task Types and Parameters

All task types support the common fields listed in the Task Configuration section above (in the Metadata Schema section). Below are the task-type-specific parameters:

1. Notebook Tasks (task_type: "notebook")

Required Parameters:

• file_path: Path to the notebook file (e.g., /Workspace/Users/[e-mail]/my_notebook)

Optional Parameters:

• parameters: Dictionary of parameters passed to the notebook as base_parameters
  • Only user-defined parameters from metadata are passed
  • If your notebook uses widgets (e.g., task_key, control_table), Databricks Jobs UI automatically adds them

Example:

- task_key: "ingest_data"
  task_type: "notebook"
  file_path: "/Workspace/Users/[e-mail]/ingestion_notebook"
  parameters:
    catalog: "my_catalog"
    schema: "my_schema"
    source_table: "source_table"

Note: If your notebook defines widgets like dbutils.widgets.text("task_key", "default"), the Databricks Jobs UI will automatically populate them when the notebook runs as a job task. You don't need to include these in the parameters field.

2. SQL Query Tasks (task_type: "sql_query")

Required Parameters:

• warehouse_id: SQL warehouse ID (can be provided via default_warehouse_id in orchestrator)
• Either sql_query OR query_id:
  • sql_query: Inline SQL query string
  • query_id: ID of a saved query in Databricks SQL

Optional Parameters:

• parameters: Dictionary of parameters for SQL parameter substitution (:parameter_name)

Example:

- task_key: "validate_data"
  task_type: "sql_query"
  warehouse_id: "abc123"
  sql_query: "SELECT COUNT(*) as row_count FROM :catalog.:schema.customers"
    parameters:
    catalog: "my_catalog"
    schema: "my_schema"

3. SQL File Tasks (task_type: "sql_file")

Required Parameters:

• warehouse_id: SQL warehouse ID (can be provided via default_warehouse_id in orchestrator)
• file_path: Path to the SQL file (e.g., /Workspace/Users/[e-mail]/query.sql)

Optional Parameters:

• parameters: Dictionary of parameters for SQL parameter substitution (:parameter_name)

Example:

- task_key: "transform_data"
  task_type: "sql_file"
  warehouse_id: "abc123"
  file_path: "/Workspace/Users/[e-mail]/transformations.sql"
    parameters:
    catalog: "my_catalog"
    schema: "my_schema"

4. Python Wheel Tasks (task_type: "python_wheel")

Required Parameters:

• package_name: Name of the Python wheel package
• entry_point: Entry point function name in the package

Optional Parameters:

• parameters: List of parameters to pass to the entry point function (can be a list or dict)

Example:

- task_key: "run_python_wheel"
  task_type: "python_wheel"
  package_name: "my_package"
  entry_point: "main"
  parameters: ["arg1", "arg2", "arg3"]

5. Spark JAR Tasks (task_type: "spark_jar")

Required Parameters:

• main_class_name: Fully qualified name of the main class (e.g., com.example.MainClass)

Optional Parameters:

• parameters: List of parameters to pass to the main class (can be a list or dict)

Example:

- task_key: "run_spark_jar"
  task_type: "spark_jar"
  main_class_name: "com.example.MainClass"
  parameters: ["param1", "param2"]

6. Pipeline Tasks (task_type: "pipeline")

Required Parameters:

• pipeline_id: ID of the Lakeflow Declarative Pipeline to run

Example:

- task_key: "run_pipeline"
  task_type: "pipeline"
  pipeline_id: "1165597e-f650-4bf3-9a4f-fc2f2d40d2c3"

7. dbt Tasks (task_type: "dbt")

Required Parameters:

• commands: dbt command to execute (e.g., "dbt run --models my_model")
• project_directory: Path to dbt project directory. Optional for Git sourced tasks, in which case if no value is provided, the root of the Git repository is used.

Optional Parameters:

• profiles_directory: Relative path to the profiles directory. Can only be specified if no warehouse_id is specified. If no warehouse_id is specified and this folder is unset, the root directory is used.
• warehouse_id: SQL warehouse ID (can be provided via default_warehouse_id in orchestrator, which is used as default if not specified)
• catalog: Catalog name for dbt
• schema: Schema name for dbt

Example:

- task_key: "run_dbt"
  task_type: "dbt"
  commands: "dbt run --models my_model"
  warehouse_id: "abc123"
  profiles_directory: "/Workspace/Users/[e-mail]/dbt-profiles"
  project_directory: "/Workspace/Users/[e-mail]/dbt-project"
  catalog: "main"
  schema: "analytics"

Task Disabling

The disabled field in the control table controls whether a task is disabled in the Databricks job:

• disabled: false → Task is enabled (default)
• disabled: true → Task is disabled

Disabled tasks are included in the job definition but will not execute when the job runs. This allows you to temporarily disable tasks without removing them from the job. Dependencies on disabled tasks are still satisfied (the task exists, it just doesn't execute).

Audit Fields

Both the control table and jobs table automatically track user information:

• created_by: Username of the user who created the record (set only on INSERT, never updated)
• updated_by: Username of the user who last updated the record (set on INSERT and UPDATE)
• created_timestamp: Timestamp when the record was created
• updated_timestamp: Timestamp when the record was last updated

These fields are automatically populated using Spark SQL's current_user() function.

Task Dependencies

Tasks use the depends_on field to specify dependencies. Tasks without dependencies (or with empty depends_on) run first in parallel. Tasks with dependencies wait for all their dependencies to complete before running.

Example:

tasks:
  - task_key: "task_a"
    depends_on: []  # No dependencies, runs first
  - task_key: "task_b"
    depends_on: []  # No dependencies, runs in parallel with task_a
  - task_key: "task_c"
    depends_on: ["task_a", "task_b"]  # Waits for both task_a and task_b to complete

The framework automatically resolves dependencies and creates the correct execution order. Circular dependencies are detected and will cause an error.

Examples

See examples/metadata_examples.yaml for comprehensive examples including:

• Data quality checks
• Bronze to Silver transformations
• Mixed task type pipelines
• End-to-end data pipelines

Error Handling and Troubleshooting

Job-Level Error Isolation (Single File)

When loading a single YAML file, the framework handles errors at the job level, ensuring that one invalid job doesn't prevent other jobs from being processed:

Variable Substitution Errors:

Failed to load job 'My Job' (resource_id='job_${var.undefined}'): 
  Variable substitution failed: Variable '${undefined}' is referenced 
  but not provided in variables dict. Available variables: ['env', 'catalog']

Failed to load 1 job(s). Continuing with 5 valid job(s).

Common Error Scenarios (single file):

1. Undefined variable: Job is skipped, other jobs continue
2. Invalid task type: Job is skipped, other jobs continue
3. Circular dependencies: Job is skipped, other jobs continue
4. Missing required fields: Job is skipped, other jobs continue

Atomic Multi-File Loading (Folder/Volume)

When loading multiple YAML files using load_from_folder() or sync_from_volume(), the loading process is atomic:

All-or-Nothing Behavior:

• ✅ Phase 1: ALL files are parsed and validated in memory
• ✅ Phase 2: Only if ALL files are valid, a single database write occurs
• ❌ If ANY file has an error, NO data is written to the control table
• 📋 Detailed error message indicates which file caused the failure

Example Error (Multi-File):

Failed to load YAML files from folder '/Workspace/metadata/'. 
Error in file '/Workspace/metadata/invalid_job.yaml': 
  Duplicate resource_id 'etl_pipeline' found across multiple YAML files. 
  This resource_id appears in 'invalid_job.yaml' and was already defined in a previous file. 
  Each resource_id must be unique across all YAML files. 
No data has been loaded to the control table.

Why Atomic Loading?

• Prevents partial data corruption (some files loaded, others failed)
• Ensures consistency across your job definitions
• Makes debugging easier (all files must be valid together)
• Protects against duplicate resource_ids across files

Validation Errors That Stop Multi-File Loading:

1. Duplicate resource_id across files: Each resource_id must be unique globally
2. Invalid YAML syntax: Any file with malformed YAML
3. Validation errors: Missing required fields, circular dependencies, etc.
4. Variable substitution errors: Undefined variables in ANY job

Best Practice for Multi-File Loading:

try:
    tasks, resource_ids = jm.load_from_folder(
        folder_path="/Workspace/metadata/",
        var={'env': 'prod', 'catalog': 'bronze'}
    )
    print(f"✅ Successfully loaded {tasks} tasks from {len(resource_ids)} jobs")
except RuntimeError as e:
    print(f"❌ Failed to load YAML files: {e}")
    # Fix the problematic file and retry
    # No partial data was written

Checking Results

Single File Loading:

jobs = jm.create_or_update_jobs(
    yaml_path="/path/to/metadata.yaml",
    control_table="catalog.schema.control",
    var=vars
)

# jobs contains list of successfully processed jobs
# Check logs for warnings about skipped jobs

Multi-File Loading:

try:
    jobs = jm.create_or_update_jobs(
        yaml_path="/Workspace/metadata/",  # folder or volume
        control_table="catalog.schema.control",
        var=vars
    )
    # All files loaded successfully
    print(f"✅ Loaded {len(jobs)} jobs")
except RuntimeError as e:
    # No files were loaded
    print(f"❌ Failed: {e}")

Variable Substitution Best Practices

1. Always define all variables: Ensure all referenced variables are provided in the var dict
2. Use descriptive variable names: ${var.environment} instead of ${var.e}
3. Validate variables before use: Check that your var dict contains all required keys
4. Use correct syntax: ${var.name} with curly braces, not $(var.name) with parentheses
5. Test with variable values: Test your YAML with actual variable values before deployment

Example:

# Define all variables upfront
vars = {
    'env': 'prod',
    'source': 'customers',
    'warehouse_id': 'abc123',
    'catalog': 'bronze',
    'schema': 'raw'
}

# Validate variables match what's in YAML
required_vars = ['env', 'source', 'warehouse_id', 'catalog', 'schema']
missing = [v for v in required_vars if v not in vars]
if missing:
    raise ValueError(f"Missing required variables: {missing}")

# Now load jobs
jobs = jm.create_or_update_jobs(
    yaml_path="/path/to/template.yaml",
    control_table="catalog.schema.control",
    var=vars
)

Best Practices

1. Use SQL Files for Reusable Logic: Store common SQL transformations in files
2. Use Task Dependencies: Design your pipelines with clear dependencies using depends_on
3. Parameterize SQL: Use parameters and Databricks dynamic value references for flexible SQL
4. Follow Naming Conventions: Use clear, descriptive task_key values
5. Document Your Pipelines: Add descriptions to jobs
6. Use Task Parameters: Leverage Databricks dynamic value references like {{job.id}}, {{task.name}}, {{job.start_time.iso_date}} for runtime values
7. Unified Parameters: Use the parameters field for all task parameters instead of separate source_config/target_config

Future Enhancements

• PowerBI refresh Tasks
• Enhanced error handling and retry logic
• Execution monitoring dashboard
• Data quality framework
• Databricks Apps

Contributing

Contributions are welcome! Please open an issue or submit a pull request on the GitHub repository.

For questions, bug reports, or feature requests, please open an issue.