create_auto_cdc_from_snapshot_flow Python session resolution fails if having multiple snapshot flows

[david_aspegren]

When a pipeline contains more than one create_auto_cdc_from_snapshot_flow flow (each driven by a custom Python next_snapshot_and_version function), flow resolution fails intermittently/consistently with:

RuntimeError: The original Spark session is being accessed instead of the per-flow cloned session during parallel analysis. This is commonly caused by spawning threads inside a flow function that access the Spark session.

I am having functions to figure out the next snapshot like this:
```
def next_x_snapshot_and_version(latest_version):
versions = spark.read.table(SOURCE_TABLE).select("file_modification_time").distinct()
```

Is this a bug?

[ThomasBehne]

This error occurs because your custom `next_x_snapshot_and_version` function references the global `spark` session variable directly, bypassing Delta Live Tables' per-flow cloned session during parallel execution. When running multiple CDC snapshot flows concurrently, DLT isolates each flow using its own cloned session; accessing global `spark` state breaks this thread safety. To fix it, update your function signature to accept a `spark_session` argument explicitly (e.g., `def next_x_snapshot_and_version(spark_session, latest_version) and pass the thread-safe `spark` instance into your custom function via a `lambda` inside your `create_auto_cdc_from_snapshot_flow` call.

[david_aspegren]

Thank you! I am still doing something wrong though, my function is now called like so:

dp.create_auto_cdc_from_snapshot_flow(

    target=SILVER_TABLE,

    source=lambda latest_version: next_locations_snapshot_and_version(spark, latest_version),

    keys=["location_id"],

    stored_as_scd_type=2,

    track_history_except_column_list=["file_modification_time", "source_file", "ingestion_time"],

)

anything more i need to do to get to "pass the thread-safe `spark` instance"

[Satyasai]

Try this
In your next_x function

def next_x_snapshot_and_version(latest_version):
versions = spark.read.table(SOURCE_TABLE).select("file_modification_time").distinct()
Replace with 
def next_x_snapshot_and_version(latest_version):
active_spark = SparkSession.getActiveSession()
versions = active_spark .read.table(SOURCE_TABLE).select("file_modification_time").distinct()

In this Below code, Remvoe Blod code, lambda latest_version:, directly call the function

dp.create_auto_cdc_from_snapshot_flow(

    target=SILVER_TABLE,

    source=lambda latest_version: next_locations_snapshot_and_version(spark, latest_version),

    keys=["location_id"],

    stored_as_scd_type=2,

    track_history_except_column_list=["file_modification_time", "source_file", "ingestion_time"],

)

[srini_ve]

@david_aspegren 
I don't think this is necessarily a bug. It looks more like an issue with how the custom next_snapshot_and_version() function is being evaluated when multiple create_auto_cdc_from_snapshot_flow flows are analysed in parallel.

In your example:

def next_x_snapshot_and_version(latest_version):
versions = (
spark.read.table(SOURCE_TABLE)
.select("file_modification_time")
.distinct()
)

the function is directly accessing the global spark session. During parallel flow analysis, Lakeflow uses a cloned Spark session for each flow, so reaching back to the original spark session can cause the error you're seeing.

I would make the next_snapshot_and_version() function a pure Python function, no spark, dbutils, table reads, or other external state inside it.

For example, if the next version is simply based on the current version:

def next_snapshot_and_version(latest_version):
next_version = latest_version + 1
next_snapshot = f"snapshot_{next_version}"

return next_snapshot, next_version

If you need to choose the next snapshot from a list, you can still keep the function pure:

def next_snapshot_and_version(latest_version, available_snapshots):
candidates = [
s for s in available_snapshots
if s["version"] > latest_version
]

if not candidates:
return None

next_snapshot = min(
candidates,
key=lambda x: x["version"]
)

return next_snapshot["path"], next_snapshot["version"]

The important part is that available_snapshots should be obtained outside the function, in the appropriate pipeline/flow context. The function itself only works with the values passed to it.

So instead of doing this:

def next_snapshot_and_version(latest_version):
## Spark access inside the custom function
df = spark.read.table(SOURCE_TABLE)

I would use this pattern:

##### Spark/table lookup happens in the appropriate context
available_snapshots = ...

##### Pure Python function does the calculation
next_snapshot, next_version = next_snapshot_and_version(
latest_version,
available_snapshots
)

I would also avoid using shared global variables or mutable state between the different CDC flows. Each flow should ideally have its own configuration and be independently resolvable.

So, in short, the key change I would make is:

Keep next_snapshot_and_version() Spark-free.

[david_aspegren]

thanks guys, what i have now is:

_snapshot_versions = sorted(

    row[0]

    for row in spark.read.table(SOURCE_TABLE)

                    .select(VERSION_COLUMN)

                    .distinct()

                    .collect()

)

_snapshots = {

    version: _cleanse(

        spark.read.table(SOURCE_TABLE).filter(col(VERSION_COLUMN) == lit(version))

    )

    for version in _snapshot_versions

}

def next_departments_snapshot_and_version(latest_version😞

    remaining = [

        v for v in _snapshot_versions if latest_version is None or v > latest_version

    ]

    if not remaining:

        return None

    next_version = remaining[0]

    return (_snapshots[next_version], next_version)

It works but I am yet to start validating the actual data.

@srini_ve is that close to what you were thinking?

[srini_ve]

@david_aspegren 
Yes, that’s close to what I had in mind.👍