Replications

Replications are YAML configurations that define data movement from source to target systems.

Driving replications from Python? Load the sling-python skill — the Replication class kwargs mirror these YAML keys exactly.

Gather first

Resolve each row before you write YAML, in this order: the user's request, existing files, discovery (MCP / CLI), then the default in the table. Ask the user only for rows that stay unresolved, one at a time, with a proposed default. If every row resolves, do not ask — build, and state the assumptions you took. Confirm first only for destructive resolutions (for example a mode that drops or rewrites existing tables).

When you update a replication, read the file first. Preserve stream-level overrides unless asked to drop them.

When the ask names a table but not its connection, check each DB connection with connection.discover / sling conns discover before you conclude the table does not exist. Pick the connection that has the table. Treat the table as absent only when no configured connection has it; then write the file with the requested names.

Project layout

In a Sling project, replications live in replications/ and target raw_* schemas only. Staging, intermediate, and marts are build models — not replication targets. See PROJECT.md in the sling skill.

source: MY_STRIPE
target: MY_WAREHOUSE
defaults:
  object: 'raw_stripe.{stream_table}'

Basic Structure

source: SOURCE_CONNECTION
target: TARGET_CONNECTION

defaults:
  mode: full-refresh
  object: '{target_schema}.{stream_table}'

streams:
  schema.table1:
  schema.table2:
    mode: incremental
    primary_key: [id]

env:
  SLING_THREADS: 5

Validate, then run

Validate (MCP)

{"action": "validate", "input": {"file_path": "/path/to/replication.yaml"}}

Finish with a compiled check: MCP replication validate (compile is the default). sling validate --parse-only is syntax-only. It is not evidence that the file is run-ready.

Run (CLI only)

There is no MCP run action. Execute with the CLI:

sling run -r /path/to/replication.yaml
sling run -r /path/to/replication.yaml --streams table1 -m incremental
sling run -r /path/to/replication.yaml --debug

Modes

Incremental Strategies

streams:
  orders:
    mode: incremental
    primary_key: [order_id]
    update_key: updated_at

To also capture deletes in incremental mode, set the delete_missing target option (soft marks rows, hard removes them). For true log-based change capture, use mode: change-capture — see CDC.md.

Backfill Mode

streams:
  large_table:
    mode: backfill
    update_key: created_date
    source_options:
      range: '2023-01-01,2023-12-31'
      chunk_size: 7d

Stream Configuration

Every stream must resolve an object (per stream or defaults.object). A database target without it fails compile.

streams:
  my_table:
    object: schema.target_table     # Target table
    mode: incremental               # Loading mode
    primary_key: [id]               # PK columns
    update_key: updated_at          # Incremental key
    columns: {id: bigint, amount: decimal(10,2)}
    select: [id, name, email]       # or [-password] to exclude
    sql: |                          # Custom SQL
      SELECT * FROM my_table
      WHERE status = 'active'
    disabled: false

Wildcards and Patterns

streams:
  public.*:                    # All tables in schema
  sales.customer_*:            # Tables with prefix
  public.sensitive_data:
    disabled: true             # Exclude specific
  'data/*.csv':                # All CSV files
  'logs/**/*.json':            # Recursive matching

Variables

defaults:
  object: '{target_schema}.{stream_schema}_{stream_table}'

Source Options

Database Sources

File Sources

defaults:
  source_options:
    format: csv
    header: true
    encoding: utf8

Target Options

Database Targets

File Targets

Partition a file target with part tokens in object (year, month, year_month, day, week, hour). These require update_key — a datetime column, or _sling_loaded_at (Unix epoch seconds).

Transforms

Clean, normalize, or compute columns with 50+ built-in functions — see TRANSFORMS.md for the full function reference and syntax.

streams:
  public.users:
    transforms:
      - name: 'trim_space(upper(value))'
        email: 'lower(value)'
      - full_name: 'record.first_name + " " + record.last_name'

Hooks

Run actions at lifecycle points: start/end (replication-level), pre/post (per stream), and pre_merge/post_merge (in the merge transaction). Hook types (query, check, http, log, …) are shared with pipeline steps — see ../sling-pipelines/STEPS.md for the full reference of all types, variables, and error handling.

hooks:
  end:
    - type: check
      check: execution.status.error == 0
      on_failure: break
    - type: http
      url: '{env.SLACK_WEBHOOK}'
      payload: '{"text": "Replicated {execution.total_rows} rows"}'

streams:
  public.users:
    hooks:
      post:
        - type: query
          connection: '{target.name}'
          query: "ANALYZE {object.full_name}"

A stream stage (e.g. post) replaces only the defaults.hooks of that stage. To keep the default hooks and add to them, use +stage (runs before) or stage+ (runs after): +pre, pre+, +post, post+, +pre_merge, pre_merge+, +post_merge, post_merge+.

defaults:
  hooks:
    post:
      - type: query
        query: file://log_insert.sql   # every stream

streams:
  public.orders:
    hooks:
      post+:                           # post: [log_insert, delete]
        - type: query
          query: DELETE FROM {object.full_name} WHERE log_date < CURRENT_DATE - 9

Environment Variables

env:
  SLING_THREADS: 10           # Parallel streams
  SLING_RETRIES: 3            # Retry failed streams
  SLING_LOADED_AT_COLUMN: true

Examples

Database to Database

source: POSTGRES
target: SNOWFLAKE

defaults:
  mode: incremental
  object: 'warehouse.{stream_schema}_{stream_table}'
  primary_key: [id]
  update_key: updated_at

streams:
  public.customers:
  public.orders:
  public.products:

Files to Database

source: S3
target: POSTGRES

defaults:
  mode: full-refresh
  object: 'staging.{stream_file_name}'
  source_options:
    format: csv
    header: true

streams:
  'data/customers/*.csv':
  'data/orders/*.csv':

Database to Files

source: POSTGRES
target: S3

defaults:
  mode: full-refresh
  object: 'exports/{stream_table}/{YYYY}/{MM}/'
  target_options:
    format: parquet
    compression: snappy

streams:
  public.transactions:
  public.events:

Topics Reference

More Topics (docs index)

Fetch the doc URL when a topic below comes up — this skill does not duplicate them.

Full Documentation

See https://docs.slingdata.io/concepts/replication.md for complete reference.