Trigger a full refresh in a lakeflow connect pipeline with a job

[Nmtc9to5]

Hello everyone, I'm implementing a lakeflow connect pipeline that is orchestrated by a lakeflow job. My question is: Is there a way ti configure a job parameter so that when ir receives the vale "true" a full refresh Is performed in the pipeline and in other scenarios it Performs an incremental ingestion?

[aliyasingh]

Currently, the native Pipeline task in Databricks Jobs doesn't support dynamically mapping a job parameter directly to the execution's Full Refresh flag (it operates as a static option in the UI).

However, you can easily achieve this behavior by using a lightweight Python Notebook task as an orchestrator. By leveraging the Databricks SDK, you can read the job parameter at runtime and conditionally trigger the correct update mode via the API.

Here is how you can set it up:

Step 1: Create an Orchestrator Notebook

Create a simple notebook that will act as the trigger for your LakeFlow Connect pipeline. Use the following Python code:

Python

from databricks.sdk import WorkspaceClient
import time

w = WorkspaceClient()

# 1. Define the widget to catch the job parameter (defaulting to "false")
dbutils.widgets.text("is_full_refresh", "false")

# 2. Retrieve and evaluate the parameter
is_full_refresh_param = dbutils.widgets.get("is_full_refresh").strip().lower()
run_full_refresh = is_full_refresh_param == "true"

# Replace with your actual pipeline ID
pipeline_id = "<your-lakeflow-pipeline-id>"

# 3. Start the pipeline update
print(f"Triggering pipeline {pipeline_id}. Full refresh: {run_full_refresh}")

update = w.pipelines.start_update(
    pipeline_id=pipeline_id,
    full_refresh=run_full_refresh
)

print(f"Update started successfully. Update ID: {update.update_id}")

# Optional: Add a polling loop if you want the Job task to wait for the pipeline to finish
# before moving to the next task.
# ... 

Step 2: Configure Your Job

1. Go to your Databricks Workflow and replace your native Pipeline task with a Notebook task pointing to the notebook you just created.

2. In the Parameters section of your Job or Task, add a key named is_full_refresh.

3. When you run the job with parameters, you can now pass the value "true" to force a full refresh, or "false" (or leave it as default) to perform your standard incremental ingestion.

Why this works: LakeFlow Connect pipelines are built on Delta Live Tables (DLT) architecture under the hood. The start_update method in the Databricks SDK natively interacts with the same API that the UI uses, giving you full programmatic control over the full_refresh boolean flag.

[balajij8]

You can solve it using the conditional tasks in the jobs. You can use If/Else conditional tasks in the main job configuration to run full/incremental based on the parameter. It allows you to route to different pipeline tasks based on the parameter value - all within job configuration with no code needed.

You can follow below

 

 Create main lake flow job with 3 tasks as below

1. Create an If/Else Conditional task- check_refresh_mode - Conditional task that evaluates {{job.parameters.full_refresh}} == "true"
2. Create a pipeline task run_full_refresh to run the lakeflow connect pipeline (Select the Trigger a full refresh on the pipeline) - Pipeline task runs when condition full_refresh is true with complete data reingest

    3. Create a pipeline task run_incremental to run the lakeflow connect pipeline (Dont select the Trigger a full refresh on the pipeline) - Pipeline task runs when condition full_refresh is false with incremental data ingest

   You can set a schedule for the main lakeflow job to run incremental ingest with default full_refresh as false. Run the lakeflow job manually with full_refresh as true for full reingest