Refresh Incremental Dataform Models When Data Lands

Incremental models exist to process only new rows, yet most teams still run them on a timer, so freshly landed data waits hours for the next scheduled window. This blueprint inverts that: the ingestion pipeline calls a Kestra webhook the moment its load finishes, and io.kestra.plugin.dataform.cli.DataformCLI immediately runs dataform run --tags incremental against BigQuery. Dataform's incremental table semantics insert only the new rows, the flow completes in minutes, and Slack confirms the marts caught up with the source.

How it works

1. The upstream_load_webhook trigger (io.kestra.plugin.core.trigger.Webhook) starts the flow on demand, secured by a webhook key from a secret. The final step of any ingestion pipeline, a dlt load, a Fivetran sync, a custom loader, calls it with one HTTP request.
2. run_incremental_models (io.kestra.plugin.dataform.cli.DataformCLI) loads the project from namespace files, installs the pinned @dataform/core dependency in beforeCommands, and runs dataform run --tags incremental with credentials from inputFiles and GOOGLE_APPLICATION_CREDENTIALS.
3. Models declared with type: "incremental" in their SQLX config and tagged incremental merge only rows beyond their WHERE watermark, so the run stays fast no matter how large the tables grow.
4. notify posts the completed refresh to Slack, and the errors block posts a distinct alert, an alert that matters more than most, because failing here means landed data is sitting unprocessed.

What you get

• Mart freshness measured in minutes after landing, not hours until the next cron window.
• Cheap runs, incremental semantics plus tag scoping touch only new rows in the affected models.
• A clean contract between ingestion and transformation, one webhook call, no shared scheduler state.
• An alert that fires precisely when landed data is not being processed.

Who it's for

• Data engineers whose ingestion finishes at unpredictable times and whose transformations should follow immediately.
• Analytics teams cutting end-to-end latency between source systems and dashboards without a streaming rewrite.
• Anyone who has explained to a stakeholder why data loaded at 09:10 appears in the dashboard at 14:00.

Why orchestrate this with Kestra

Chaining transformation to ingestion by timing guesses, run dataform an hour after the load usually finishes, breaks the first time the load runs long. Kestra makes the dependency explicit: the webhook is a stable URL any loader can call, every refresh is an execution with logs tied to the load that caused it, and the same flow can grow gates, notifications, and downstream triggers without touching the ingestion side.

Prerequisites

• A Dataform project stored as namespace files, with incremental models tagged incremental in their config blocks.
• An upstream pipeline able to send one HTTP request when its load completes.
• A BigQuery service account with permission to run jobs and write to the target datasets.
• A Slack incoming webhook for run reports.

Secrets

• GCP_SERVICE_ACCOUNT_JSON: the service account key JSON, delivered to the task as sa.json.
• SLACK_WEBHOOK_URL: Slack incoming webhook URL.
• WEBHOOK_KEY: shared key that authorizes calls to the webhook trigger.

Quick start

1. Add the GCP_SERVICE_ACCOUNT_JSON, SLACK_WEBHOOK_URL, and WEBHOOK_KEY secrets to your Kestra namespace.
2. Upload your Dataform project to the namespace files and tag the incremental models.
3. Call the webhook URL with curl and check that Slack reports the finished refresh.
4. Add the webhook call as the final step of your ingestion pipeline.

How to extend

• Trigger from another Kestra flow with a Flow trigger on its SUCCESS state instead of a webhook, keeping the whole chain inside Kestra.
• Gate the refreshed models on assertions before dashboards read them, as shown in the assertions gate blueprint.
• Pair with a periodic full rebuild to correct incremental drift, as shown in the full refresh weekend blueprint.
• Pass the upstream table name in the webhook body and select the matching tag from {{ trigger.body }} for per-source refreshes.

Links

• Dataform plugin documentation
• Dataform documentation