Block 1 (AECTM):

@dlt.view(

comment=comment if comment is not None else f"define_key({key_expr})",

name=target_table_name

)

def key_func():

df = None

if '://' in topic_name:

if streaming:

df = spark.readStream.format("delta").option("mergeSchema", "true").option("ignoreChanges", "true").load(topic_name)

else:

df = spark.read.format("delta").option("mergeSchema", "true").option("ignoreChanges", "true").load(topic_name)

else:

if streaming:

df = dlt.readStream(topic_name)

else:

df = dlt.read(topic_name)

return df.withColumn("key", eval(key_expr))

Block 2&3 (sequence & scd_type1):

target_table_name = alias if alias is not None else f"scd_type1({topic_name},{sequence_col})"

dlt.create_streaming_live_table(

comment=comment if comment is not None else f"scd_type1({topic_name},{sequence_col})",

name=target_table_name

)

@dlt.table(

comment=comment if comment is not None else f"sequence({topic_name})",

name=f"sequence({topic_name})"

)

@dlt.expect_or_drop("null key", col("key").isNotNull())

def seq_view():

if streaming:

return dlt.readStream(topic_name).withColumn("sequence", eval(sequence_col))

else:

return dlt.read(topic_name).withColumn("sequence", eval(sequence_col))

dlt.apply_changes(

target=target_table_name,

source=f"sequence({topic_name})",

keys=["key"],

sequence_by=col("sequence"),

stored_as_scd_type="1"

)

Block 4 (filter view):

target_table_name = alias if alias is not None else f"filter({filter_cond})"

@dlt.view(

comment=comment if comment is not None else f"filter({filter_cond})",

name=target_table_name

)

def flt():

df = dlt.readStream(topic_name) if streaming else dlt.read(topic_name)

return df.filter(eval(filter_cond))

Block 5 (MAIN_FLOW_MOVES):

target_table_name = alias if alias is not None else f"main_flow_moves({topic_name})"

def getn(position_info, level):

if position_info is not None:

if position_info.level == level:

return position_info.name

else:

return getn(position_info.parent, level)

else:

return None

udfGetName = udf(getn)

@dlt.table(

comment=comment if comment is not None else f"main_flow_moves({topic_name})",

name=target_table_name

)

def func():

df = dlt.readStream(topic_name) if streaming else dlt.read(topic_name)

dfSeries = df.select("containerTerminalVisitKey", col('moveEvents')[0]['from']['location']['locationType'].alias("fromWhere"), explode('moveEvents.to').alias("toLoc")). \

select("containerTerminalVisitKey", (col("toLoc.time") / 1000.0).cast(TimestampType()).alias("time"), "fromWhere", "toLoc.location.positionInfo"). \

withColumn("blockName", udfGetName("positionInfo", lit("BLOCK"))). \

withColumn("bayName", udfGetName("positionInfo", lit("BAY"))). \

withColumn("slotName", udfGetName("positionInfo", lit("STACK"))). \

filter(col("blockName").isNotNull() & col("bayName").isNotNull() & col("slotName").isNotNull()). \

drop("positionInfo")

return dfSeries

==================================================

The input dataframe schema is actually quite intimidating with recursive struct definitions, but most of it is just taken along for the ride. The key field is containerTerminalVisitKey and the sequence field is containerInfo.updateCounter. The latter is just a sequence number (long).

I think there is a lot going on. I'm not sure how much the code that I shared helps. I was thinking that understanding the situation in which the error message (in the original post) is emitted might help.