我在pyspark中有一个框架，我每天都读同一个文件。新的数据被添加到这个文件中。我把数据插入到一个delta表中。
我想更新已经插入的数据，因为有时数据会在我读取的文件中更新。
我的网络看起来像这样：

col1    col2    col3           col4
name_1  city_1  date_of_today  {'Date': X, "Weight (KG)": 13.4}
name_1  city_1  date_of_today  {'Date': X, "Weight (KG)": 1000}

字符串
我把它插入到delta表中，我想更新满足条件的数据。
我这样做了：

from delta.tables import DeltaTable
deltaTable = DeltaTable.forPath(spark, "abfss://[email protected]/db/db_name/table/")

deltaTable.alias("target").merge(
    df_content_transformed.alias("source"),
    condition
    ) \
.whenMatchedUpdateAll() \
.whenNotMatchedInsertAll() \
.execute()

型
条件值为：

"""target.name = source.name AND target.city= source.city AND get_json_object(target.value, '$.Date') = get_json_object(source.value, '$.Date') AND get_json_object(target.value, '$.Weight (KG)') = get_json_object(source.value, '$.Weight (KG)')"""

型
当我执行一个管道调用这个代码时，我有以下错误：

Py4JJavaError: An error occurred while calling o4197.execute.
: org.apache.spark.sql.delta.DeltaUnsupportedOperationException: Cannot perform Merge as multiple source rows matched and attempted to modify the same
target row in the Delta table in possibly conflicting ways. By SQL semantics of Merge,
when multiple source rows match on the same target row, the result may be ambiguous
as it is unclear which source row should be used to update or delete the matching
target row. You can preprocess the source table to eliminate the possibility of
multiple matches.

型
我能做什么？