UPDATE:
See the code below (for some reason couldn't post it yesterday.)
I decided to revert back to an earlier version of the DLT pipeline which defined each table one by one rather than programatically, which had run without issue. That ran fine.
I added on a few extra tables and ran it before signing off last night, and started the job that updates the source tables
Today I ran the DLT pipeline (and adds to the source tables) and the second of the three tables/steps that I added failed with the same error.
@dataclass
class IdTable:
schema_name: str
table_name: str
email_col_name: str
idno_col_name: Optional[str] = None
surname_col_name: Optional[str] = None
table_user_id_col: Optional[str] = None
def generate_raw_table(source_path: str, target_name_base: str):
@dlt.table(
name=f"{target_name_base}_raw" ,
comment=f"Raw records from bronze layer for {source_path}"
)
def create_live_raw_table():
return (spark.readStream.table(source_path))
def generate_cleaned_bronze_table(target_name_base: str, id_table: IdTable):
@dlt.table(
name= f"{target_name_base}_clean"
, comment=f"Cleaned records from raw bronze views for {target_name_base}_raw"
# , partition_cols=["insert_time", "ref", "table_uid"]
)
def create_live_clean_table():
df_raw = spark.readStream.table(f"LIVE.{target_name_base}_raw")
df_clean = df_raw.withColumn(
# ...
)
return df_clean
id_tables: List[IdTable] = [
IdTable(
schema_name="source_name",
table_name="acc_user_account",
idno_col_name="identification_nr",
surname_col_name="surname",
email_col_name="email_address",
table_user_id_col="user_id"
),
]
for table in id_tables:
source_path: str = f"bronze_raw.{table.schema_name}.{table.table_name}"
target_name_base: str = f"{table.schema_name}_{table.table_name}"
generate_raw_table(source_path=source_path, target_name_base=target_name_base)
generate_cleaned_bronze_table(target_name_base=target_name_base, id_table=table)
the downstream code, which worked both runs
@dlt.table(
comment = "UUIDs for where name AND id AND email match",
partition_cols=["ref", "deduped_uid_1"],
)
def deduped_1_name_id_email():
q: str = generate_dedup_query(
source_table='deduped_0_per_table',
matching_cols=['ref'],
source_id_col='deduped_uid_0',
output_id_col='deduped_uid_1'
)
return spark.sql(q)
the table that worked on the first run, then errored on the second
@dlt.table(
comment = "UUIDs for where surname AND id NOT email match",
partition_cols=["idno", "surname", "deduped_uid_2"],
)
def deduped_2_name_id():
q: str = generate_dedup_query(
source_table='deduped_1_name_id_email',
matching_cols=['idno', "surname"],
source_id_col='deduped_uid_1',
output_id_col='deduped_uid_2',
where_clause="WHERE idno is not null AND surname is not null"
)
print(q)
return spark.sql(q)
My only thought is that its the values themselves causing the error, but during my debugging yesterday I tried adding a terminating column (containing a string like 'row_end') to no efffect.
I've tried storing them under diferent schemas, deleting and recreating the pipeline, no change.
