OrbexDB Jobs & Pipelines

Create a DAG with orbex.pipeline.yml

orbex.pipeline.yml is the source-controlled repository manifest for Genorbex jobs. It defines Python, PySpark, Jupyter notebook, and SQL tasks plus their dependencies, retries, runtime settings, data inputs, and outputs.

Download example YAML JSON Schema

Git deployed

Deploy from GitHub or GitLab and synchronize repository task files.

Dependency aware

dependsOn builds the visual DAG, lineage, parallel branches, and execution order.

Governed data

Read and write OrbexDB, Orbex Buckets, AWS S3, GCS, and Azure Blob.

Quickstart

1. Add job files. Put Python files under a repository folder such as jobs/.
2. Add the manifest. Save orbex.pipeline.yml at the repository root.
3. Push the branch. Commit the manifest and every referenced job file to GitHub or GitLab.
4. Deploy. Open OrbexDB → Jobs & Pipelines → Deploy from Git, authenticate, and select the repository. OrbexDB detects its branches and tags and selects the default branch automatically.

orbex.pipeline.yml

version: 1

defaults:
  # Fail quickly instead of occupying a worker for repeated long attempts.
  maxRetries: 0
  timeoutSeconds: 300

tasks:
  - id: generate-mock-source
    type: PYTHON
    path: jobs/generate_mock_data.py

  - id: spark-clean-customers
    type: PYSPARK
    path: jobs/spark_clean_customers.py
    dependsOn: [generate-mock-source]
    sparkConfig:
      spark.executor.memory: 2g
      spark.executor.cores: "1"

  - id: spark-enrich-customers
    type: PYSPARK
    path: jobs/spark_enrich_customers.py
    dependsOn: [spark-clean-customers]
    sparkConfig:
      spark.executor.memory: 2g
      spark.executor.cores: "1"

  - id: spark-aggregate-customers
    type: PYSPARK
    path: jobs/spark_aggregate_customers.py
    dependsOn: [spark-enrich-customers]
    sparkConfig:
      spark.executor.memory: 2g
      spark.executor.cores: "1"

  - id: spark-validate-customers
    type: PYSPARK
    path: jobs/spark_validate_customers.py
    dependsOn: [spark-enrich-customers]
    sparkConfig:
      spark.executor.memory: 2g
      spark.executor.cores: "1"

  - id: ingest-customers
    type: PYTHON
    path: jobs/ingest_customers.py
    dependsOn: [spark-validate-customers]

  - id: publish-metrics
    type: PYTHON
    path: jobs/publish_ingestion_metrics.py
    dependsOn: [ingest-customers, spark-aggregate-customers]
    output:
      type: TABLE
      databaseName: analytics
      schemaName: PUBLIC
      tableName: pipeline_metrics
      mode: APPEND

YAML indentation matters. Spark keys must be nested under sparkConfig, as shown above. Replace analytics with an existing OrbexDB database name; it is validated during deployment. Paths are relative to the repository root and are case-sensitive.

Manifest field reference

Field	Required	Description
version	Yes	Manifest version. Use 1.
defaults.maxRetries	No	Default retry count for every task.
defaults.timeoutSeconds	No	Default task timeout in seconds, capped at one hour.
tasks	Yes	Non-empty list of DAG tasks.
tasks[].id	Yes	Stable unique task ID referenced by dependsOn.
tasks[].type	No	PYTHON, PYSPARK, JUPYTER, or SQL. Defaults to PYTHON.
tasks[].path	File jobs	Repository-relative .py or .ipynb path.
tasks[].code	Inline jobs	Inline Python or SQL source instead of path.
tasks[].dependsOn	No	Upstream task IDs. Independent tasks may run in parallel.
tasks[].sparkConfig	PySpark	Spark configuration map. Values may be strings or numbers.
tasks[].parameters	No	JSON-compatible values exposed as orbex_parameters.
tasks[].maxRetries	No	Task-specific retry override.
tasks[].retryDelaySeconds	No	Delay between retry attempts.
tasks[].timeoutSeconds	No	Task-specific timeout in seconds.
tasks[].databaseName	SQL/table jobs	Recommended OrbexDB database name; validated during Git deployment.
tasks[].databaseId	No	Internal database ID. Prefer databaseName in repository manifests.
tasks[].input	No	Governed AWS S3, GCS, or Azure Blob input.
tasks[].output	No	OrbexDB table, Orbex Bucket, S3, GCS, or Azure Blob output.
tasks[].output.databaseName	Table output	Name of an existing accessible OrbexDB database.
tasks[].output.mode	No	APPEND, OVERWRITE, TRUNCATE, UPDATE, MERGE, or UPSERT. Defaults to APPEND.
tasks[].output.keyColumns	UPDATE/MERGE	One or more existing table columns that uniquely identify a row.

Python and PySpark data contract

Every Python-family task receives orbex_parameters and orbex_input_rows. Assign orbex_output_rows or orbex_output_df to pass structured data to downstream tasks and the configured output destination.

jobs/transform.py

# Rows returned by upstream tasks or a configured cloud input.
source_rows = orbex_input_rows

# Return structured rows to downstream tasks and the configured destination.
orbex_output_rows = [
    {"customer_id": row["customer_id"], "status": "ready"}
    for row in source_rows
]

# Optional governed SQL executed by OrbexDB after the worker returns.
orbex_sql_statements = [
    "INSERT INTO PUBLIC.pipeline_audit VALUES ('customer-job', CURRENT_TIMESTAMP)"
]

jobs/aggregate.py (PySpark)

source_df = spark.createDataFrame(orbex_input_rows)

orbex_output_df = (
    source_df
    .filter("customer_id IS NOT NULL")
    .groupBy("region")
    .count()
)

OrbexDB table write modes

Table outputs auto-create a missing schema and table from the first non-empty result. Every execution log records the resolved mode, incoming row count, and data files written.

Mode	Behavior	Keys
APPEND	Adds every returned row while preserving existing table files and rows.	Not required
OVERWRITE	Truncates the table and then inserts the returned rows.	Not required
TRUNCATE	Removes every existing row and ignores returned rows.	Not required
UPDATE	Updates matching rows only. Unmatched incoming rows are ignored.	Required
MERGE / UPSERT	Updates matching rows and inserts unmatched incoming rows.	Required

Append rows

output:
  type: TABLE
  databaseName: analytics
  schemaName: PUBLIC
  tableName: sales_training_data
  mode: APPEND

Update sales rows by composite key

output:
  type: TABLE
  databaseName: analytics
  schemaName: PUBLIC
  tableName: sales_training_data
  mode: UPDATE
  keyColumns: [date, store_id, product_id]

UPDATE, MERGE, and UPSERT require unique, non-null keyColumns. Keys are matched case-insensitively by column name. Managed output modes support tables up to 100,000 rows; use a native SQL MERGE for larger tables.

Cloud storage input and output

Authenticate AWS S3, Google Cloud Storage, or Azure Blob through an Orbex Connector or Secret Handler. Credentials remain in OrbexCloud and are never sent to job code. Inputs support CSV, JSON, XLSX, and XLS; structured cloud outputs support JSON and CSV.

Cross-cloud pipeline task

tasks:
  - id: summarize-orders
    type: PYSPARK
    path: jobs/summarize_orders.py
    input:
      type: AWS_S3
      connectorInstanceId: your-s3-connector-id
      bucketName: inbound-orders
      path: daily/
      format: csv
      mode: REPLACE
      filePattern: "*.csv"
    output:
      type: GCS
      connectorInstanceId: your-gcs-connector-id
      bucketName: governed-analytics
      path: order-summary/
      fileName: summary.json
      format: json
      mode: APPEND
      partitionColumns: region

Deploy and operate

Deploy from Git

Connect GitHub or GitLab and choose an available repository. OrbexDB detects the default branch, branches, and tags automatically; use orbex.pipeline.yml as the DAG definition path.

Run now

Starts a DAG execution. Tasks without dependencies begin first; independent branches can execute in parallel.

Runs, logs, and graphs

Open a run and select a task in the lineage graph to inspect logs and returned rows. Generated PNG, JPEG, GIF, WebP, and SVG charts appear as viewable and downloadable graph artifacts.

Automatic Git sync

Keep Git synchronization enabled when deploying. Signed pushes to the selected branch reload the manifest and referenced files into a new pipeline version without starting a run.

Validation and troubleshooting

Every dependsOn value must name an existing task; cycles are rejected.
Every path must exist on the selected Git branch. Uncommitted local files are not deployed.
Use spaces, not tabs, and indent properties below sparkConfig, input, and output.
Production Python and PySpark jobs require the integrated OrbexCloud pipeline runtime to be healthy.
Set a realistic timeout. A timeout is a hard task failure and downstream dependent tasks are skipped.
If UPDATE changes no rows, inspect the task log for the matched-row count and verify that every returned row contains the configured keyColumns.