[Python] Delta lake MERGE error: [INVALID_EXTRACT_BASE_FIELD_TYPE]

I'm trying to implement the Pyspark code below to read delta files saved in the data lake (delta_table) and join with data frame with updated records (novos_registros).

#5. Build the matching condition for the MERGE
condicao_correspondencia = " AND ".join([f"t.{col} = s.{col}" for col in chave_unica])
# 6. Use the MERGE command to update or insert data
delta_table = DeltaTable.forPath(spark, delta_path)

delta_table.alias("t").merge(
novos_registros.alias("s"),
condicao_correspondencia # Dynamically generated match condition
).whenMatchedUpdate(set={
**{col: f"s.{col}" for col in novos_registros.columns} # Update all fields of novos_registros
}).whenNotMatchedInsert(values={
**{col: f"s.{col}" for col in novos_registros.columns} # Insere todos os campos de novos_registros
}).execute()


However, I'm having problems with field incompatibility. It says that a certain field in delta_lake is in complex format, and the same is string format in the data frame. This is the error that I get.

AnalysisException: [INVALID_EXTRACT_BASE_FIELD_TYPE] Can't extract a value from "0NET_PRICE". Need a complex type [STRUCT, ARRAY, MAP] but got "STRING".


Apparently, DeltaTable.forPath(spark, delta_path) is inferring the format of the fields, while the data frame fields are all in string format. This is a generic script and I don't want to fix the fields formats into the code. It should be dynamic.

Thanks in advance.

Continue reading...