Delta Lake Merge -Fehler: [Invaly_extract_base_field_type] ⇐ Python

[Anonymous]

Ich versuche, den folgenden PYSPark -Code zu implementieren, um Delta -Dateien zu lesen, die im Data Lake (Delta_table) gespeichert sind, und mit Data Frame mit aktualisierten Datensätzen (novos_registros) zusammenzuschließen. < /p>
#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()
< /code>
Ich habe jedoch Probleme mit der Feldkompatibilität. Es heißt, dass ein bestimmtes Feld in Delta_Lake im komplexen Format ist und das gleiche Zeichenformat im Datenrahmen ist. Dies ist der Fehler, den ich erhalte. < /P>
AnalysisException: [INVALID_EXTRACT_BASE_FIELD_TYPE] Can't extract a value from "0NET_PRICE". Need a complex type [STRUCT, ARRAY, MAP] but got "STRING".
< /code>
Anscheinend schließt deltatable. Es sollte dynamisch sein.
Vielen Dank im Voraus.