Nightly Backup Reference

This checked-in configuration combines database selection, recovery-context fields, row checksums, object paths, and a cloud-storage output. It has not been executed end to end.

Complete pipeline

nightly-backup.yaml

# Nightly Database Backup Pipeline
# Simple, reliable replication to cloud storage for DR
#
# Use case: Nightly backup of orders and inventory tables from
# on-premise database to cloud cold storage (GCS Nearline/S3 Glacier)
#
# Key features:
# - Full table extract via SQL
# - Minimal transformation (just add metadata)
# - Compressed Parquet format for cost efficiency
# - Date-partitioned storage for easy recovery
# - Checksum verification

name: nightly-backup-orders
type: pipeline
description: Backup orders and inventory to cloud cold storage

config:
  input:
    # Sequence through multiple tables in one pipeline
    sequence:
      inputs:
        # Orders table
        - sql_select:
            driver: postgres # Or: mysql, odbc, mssql
            dsn: 'postgres://${DB_USER}:${DB_PASSWORD}@${DB_HOST}:5432/${DB_NAME}?sslmode=require'
            table: orders
            columns: ['*']
            where: "updated_at >= CURRENT_DATE - INTERVAL '1 day'"
          processors:
            - mapping: |
                root = this
                root._table = "orders"
                root._backup_type = "incremental"

        # Inventory table (full backup - smaller table)
        - sql_select:
            driver: postgres
            dsn: 'postgres://${DB_USER}:${DB_PASSWORD}@${DB_HOST}:5432/${DB_NAME}?sslmode=require'
            table: inventory
            columns: ['*']
          processors:
            - mapping: |
                root = this
                root._table = "inventory"
                root._backup_type = "full"

        # Order line items
        - sql_select:
            driver: postgres
            dsn: 'postgres://${DB_USER}:${DB_PASSWORD}@${DB_HOST}:5432/${DB_NAME}?sslmode=require'
            table: order_items
            columns: ['*']
            where: "created_at >= CURRENT_DATE - INTERVAL '1 day'"
          processors:
            - mapping: |
                root = this
                root._table = "order_items"
                root._backup_type = "incremental"

  pipeline:
    processors:
      # Step 1: Add backup metadata
      - mapping: |
          root = this
          root._backup_metadata = {
            "backup_date": now().ts_format("2006-01-02"),
            "backup_timestamp": now(),
            "source_host": env("DB_HOST").or("unknown"),
            "source_database": env("DB_NAME").or("unknown"),
            "pipeline_version": "1.0.0",
            "node_id": env("NODE_ID").or("unknown")
          }

      # Step 2: Calculate row checksum for integrity verification
      - mapping: |
          root = this
          # Create checksum from all original fields (excluding metadata)
          let data_fields = this.without("_table", "_backup_type", "_backup_metadata")
          root._checksum = $data_fields.format_json(no_indent: true).hash("md5").encode("hex")

  output:
    # Route to different paths based on table
    switch:
      cases:
        - check: this._table == "orders"
          output:
            gcp_cloud_storage:
              bucket: '${GCS_BACKUP_BUCKET}'
              path: 'backups/orders/${! meta("backup_date") }/orders-${!uuid_v4()}.parquet'
              content_type: application/octet-stream
              batching:
                count: 10000
                period: 60s
                processors:
                  - mapping: |
                      meta backup_date = this._backup_metadata.backup_date
                      root.record = this.format_json()
                  - parquet_encode:
                      default_compression: snappy
                      schema:
                        - name: record
                          type: UTF8

        - check: this._table == "inventory"
          output:
            gcp_cloud_storage:
              bucket: '${GCS_BACKUP_BUCKET}'
              path: 'backups/inventory/${! meta("backup_date") }/inventory-full-${!uuid_v4()}.parquet'
              content_type: application/octet-stream
              batching:
                count: 50000 # Inventory is full backup, larger batches
                period: 120s
                processors:
                  - mapping: |
                      meta backup_date = this._backup_metadata.backup_date
                      root.record = this.format_json()
                  - parquet_encode:
                      default_compression: snappy
                      schema:
                        - name: record
                          type: UTF8

        - check: this._table == "order_items"
          output:
            gcp_cloud_storage:
              bucket: '${GCS_BACKUP_BUCKET}'
              path: 'backups/order_items/${! meta("backup_date") }/items-${!uuid_v4()}.parquet'
              content_type: application/octet-stream
              batching:
                count: 10000
                period: 60s
                processors:
                  - mapping: |
                      meta backup_date = this._backup_metadata.backup_date
                      root.record = this.format_json()
                  - parquet_encode:
                      default_compression: snappy
                      schema:
                        - name: record
                          type: UTF8

        # Fallback for unknown tables
        - output:
            gcp_cloud_storage:
              bucket: '${GCS_BACKUP_BUCKET}'
              path: 'backups/unknown/${!this._table}/${!this._backup_metadata.backup_date}/data-${!timestamp_unix()}.json'
              content_type: application/json
              batching:
                count: 1000
                period: 30s

Adaptation boundary

Review snapshot consistency, watermarks, credentials, object encryption, retention, completeness checks, restore ordering, and destructive restore controls before execution.

Next: Troubleshoot the reference.