Snowflake Incremental MERGE Upsert with Kestra

id: snowflake-incremental-merge-upsert
namespace: company.team
description: |
  Load a batch file into Snowflake and apply it as an idempotent incremental
  upsert. The file is staged, then the delta COPY into a staging table and the
  MERGE into the target run together in one DML transaction, so a failed batch
  rolls back and the target never holds a half-applied load. Slack confirms.

inputs:
  - id: stage_file
    type: STRING
    defaults: customers_delta.csv
    description: File name to stage and load (already in Kestra internal storage or
      fetched upstream).

tasks:
  - id: fetch_delta
    type: io.kestra.plugin.core.http.Download
    description: Stand-in for the upstream extract; replace with your real source.
    uri: https://huggingface.co/datasets/kestra/datasets/raw/main/employees/employees00.csv

  - id: stage_upload
    type: io.kestra.plugin.jdbc.snowflake.Upload
    description: Upload the delta file to the target table's internal stage.
    from: "{{ outputs.fetch_delta.uri }}"
    fileName: "{{ inputs.stage_file }}"
    prefix: delta
    stageName: "@KESTRA.PUBLIC.%CUSTOMERS"
    compress: true

  - id: ensure_staging
    type: io.kestra.plugin.jdbc.snowflake.Query
    description: Create the persistent staging table once (DDL, outside the transaction).
    sql: CREATE TRANSIENT TABLE IF NOT EXISTS KESTRA.PUBLIC.CUSTOMERS_STAGING LIKE
      KESTRA.PUBLIC.CUSTOMERS;

  - id: merge_upsert
    type: io.kestra.plugin.jdbc.snowflake.Queries
    description: Reset staging, COPY the delta, and MERGE into the target as one DML
      transaction.
    transaction: true
    queryTag: "kestra-incremental-merge-{{ execution.id }}"
    sql: |
      DELETE FROM KESTRA.PUBLIC.CUSTOMERS_STAGING;
      COPY INTO KESTRA.PUBLIC.CUSTOMERS_STAGING
        FROM @KESTRA.PUBLIC.%CUSTOMERS
        FILE_FORMAT = (type = csv field_optionally_enclosed_by='"' skip_header = 1)
        PATTERN = '.*delta/.*'
        ON_ERROR = 'abort_statement';
      MERGE INTO KESTRA.PUBLIC.CUSTOMERS AS target
        USING (
          SELECT * FROM KESTRA.PUBLIC.CUSTOMERS_STAGING
          QUALIFY ROW_NUMBER() OVER (PARTITION BY CUSTOMER_ID ORDER BY UPDATED_AT DESC) = 1
        ) AS source
        ON target.CUSTOMER_ID = source.CUSTOMER_ID
        WHEN MATCHED AND source.UPDATED_AT > target.UPDATED_AT THEN UPDATE SET
          target.EMAIL = source.EMAIL,
          target.CITY = source.CITY,
          target.UPDATED_AT = source.UPDATED_AT
        WHEN NOT MATCHED THEN INSERT (CUSTOMER_ID, EMAIL, CITY, UPDATED_AT)
          VALUES (source.CUSTOMER_ID, source.EMAIL, source.CITY, source.UPDATED_AT);

  - id: notify
    type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
    description: Confirm the incremental merge completed.
    url: "{{ secret('SLACK_WEBHOOK_URL') }}"
    payload: |
      {
        "text": "Snowflake incremental merge complete for {{ inputs.stage_file }} (execution {{ execution.id }})."
      }

pluginDefaults:
  - type: io.kestra.plugin.jdbc.snowflake
    values:
      url: "{{ secret('SNOWFLAKE_URL') }}"
      username: "{{ secret('SNOWFLAKE_USERNAME') }}"
      password: "{{ secret('SNOWFLAKE_PASSWORD') }}"

triggers:
  - id: daily
    type: io.kestra.plugin.core.trigger.Schedule
    description: Run the incremental merge on a cadence. Adjust or disable as needed.
    cron: "0 4 * * *"
    disabled: true

errors:
  - id: alert_on_failure
    type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
    description: Alert Slack if the load or merge fails (the transaction rolls back).
    url: "{{ secret('SLACK_WEBHOOK_URL') }}"
    payload: |
      {
        "text": "Snowflake incremental merge failed and rolled back (execution {{ execution.id }})."
      }

Apply a batch of changed rows to a Snowflake target table as an idempotent incremental upsert, without ever leaving the target in a half-applied state. This blueprint uploads a delta file to the table's internal stage, resets a transient staging table, COPYs the delta in, and MERGEs into the target with deduplication on the business key, all inside a single Snowflake transaction. If any statement fails, the whole batch rolls back. Reruns are safe because the MERGE is keyed and the dedup keeps only the latest row per key.

How it works

io.kestra.plugin.core.http.Download (fetch_delta) pulls a sample CSV delta. Replace this with your real extract (S3, GCS, HTTP, SFTP, dbt artifact, CDC capture, etc.).
io.kestra.plugin.jdbc.snowflake.Upload (stage_upload) uploads the file to the target table's internal stage @KESTRA.PUBLIC.%CUSTOMERS under a delta/ prefix with gzip compression.
io.kestra.plugin.jdbc.snowflake.Query (ensure_staging) issues CREATE TRANSIENT TABLE IF NOT EXISTS ... LIKE CUSTOMERS as a one-off DDL outside the transaction.
io.kestra.plugin.jdbc.snowflake.Queries (merge_upsert) runs with transaction: true and a per-execution queryTag. It DELETEs the staging rows, COPYs the staged delta with ON_ERROR = 'abort_statement', then MERGEs into CUSTOMERS using QUALIFY ROW_NUMBER() OVER (PARTITION BY CUSTOMER_ID ORDER BY UPDATED_AT DESC) to dedup late or duplicate rows.
io.kestra.plugin.slack.notifications.SlackIncomingWebhook (notify) posts a confirmation; the alert_on_failure error handler posts a rollback notice if anything fails.
A daily Schedule trigger at 0 4 * * * is included but disabled by default.

What you get

Atomic incremental load: COPY and MERGE in one Snowflake DML transaction, with automatic rollback on failure.
Idempotent reruns thanks to a keyed MERGE and a QUALIFY ROW_NUMBER dedup on CUSTOMER_ID / UPDATED_AT.
Per-execution queryTag so every batch is traceable in Snowflake's QUERY_HISTORY.
Slack confirmation on success and a separate Slack alert on rollback.
Reusable pattern across any SCD2-style or last-write-wins upsert table.

Who it's for

Analytics and data platform engineers loading CDC or batch deltas into Snowflake.
Teams replacing fragile shell scripts or stored procedures with declarative, auditable pipelines.
ELT practitioners who want pre-load orchestration that dbt does not cover.

Why orchestrate this with Kestra

Snowflake Tasks can schedule SQL, but they cannot fetch a delta file from an external system, upload it to an internal stage, and coordinate a multi-statement transaction with retries, Slack alerting, and lineage in one declarative unit. Kestra triggers this flow on a schedule or an event (webhook, file arrival, upstream extract finishing), captures every statement's result, retries transient failures, exposes outputs to downstream tasks, and keeps the full YAML in Git. Pair it with namespace secrets for credentials so nothing is hardcoded.

Prerequisites

A Snowflake account with a warehouse, the target table KESTRA.PUBLIC.CUSTOMERS (keyed by CUSTOMER_ID with an UPDATED_AT column), and grants to create transient tables and write to the table's internal stage.
A Slack incoming webhook for notifications.

Secrets

SNOWFLAKE_URL: JDBC URL, for example jdbc:snowflake://<account>.snowflakecomputing.com?warehouse=COMPUTE_WH.
SNOWFLAKE_USERNAME: Snowflake user.
SNOWFLAKE_PASSWORD: Snowflake password (swap for privateKey for key-pair auth).
SLACK_WEBHOOK_URL: Slack incoming webhook URL.

Quick start

Add the secrets above to your Kestra namespace.
Create the target table and confirm the user can write to @KESTRA.PUBLIC.%CUSTOMERS.
Replace fetch_delta with your real extract and point stage_file at the resulting filename.
Adjust the table names, merge key, and column list in merge_upsert to your schema.
Run the flow manually, then enable the daily Schedule trigger when ready.

How to extend

Swap fetch_delta for io.kestra.plugin.aws.s3.Downloads, io.kestra.plugin.gcp.gcs.Downloads, or an SFTP task to pull deltas from your lake.
Replace the schedule with a Webhook or Flow trigger so the upsert fires the moment upstream extract finishes.
Add io.kestra.plugin.core.flow.ForEachItem to parallelize across multiple target tables.
Switch to key-pair auth by replacing password with privateKey (and privateKeyPassword if encrypted) in pluginDefaults.
Emit row counts via outputs.merge_upsert.outputs and assert thresholds with a follow-up Assert task.

Snowflake Incremental MERGE Upsert with a Transactional Load

How it works

What you get

Who it's for

Why orchestrate this with Kestra

Prerequisites

Secrets

Quick start

How to extend

Links

Tasks

Orchestrate with Kestra

Share this Blueprint

Create a Blueprint

New to Kestra?

Snowflake Incremental MERGE Upsert with a Transactional Load

How it works

What you get

Who it's for

Why orchestrate this with Kestra

Prerequisites

Secrets

Quick start

How to extend

Links

Tasks

Orchestrate with Kestra

Share this Blueprint

Create a Blueprint

More Data Blueprints

Build an ETL Pipeline with Airbyte Cloud and dbt Cloud for SaaS Analytics

Trigger an Airflow DAG

Build an ETL Pipeline with Parallel Airbyte Syncs and dbt Core Transformations

AI Document Data Extraction: Parse PDFs and Web Pages into Structured JSON with OpenAI

Run a Scheduled ETL Sync with Airbyte Cloud

Run Parallel ETL Ingestion with Airbyte to Sync Multiple Data Sources

New to Kestra?