Почему я получаю ошибку неоднозначности столбца в моем запросе pyspark? ⇐ Python

[Гость]

Я пытаюсь создать столбец, который определяет, какие столбцы обновляются в рамках канала данных изменений (я следовал этому stackoverflow: Сравните два фрейма данных Pyspark), но когда я ссылаюсь на свои собственные таблицы, я получаю эту ошибку:
р>
AnalysisException:Column _commit_version#203599L, subscribe_status#203595, _change_type#203598, _commit_timestamp#203600, subscribe_dt#203596, end_sub_dt#203597 неоднозначны. Возможно, это связано с тем, что вы объединили несколько наборов данных, и некоторые из этих наборов данных совпадают. Этот столбец указывает на один из наборов данных, но Spark не может определить, какой именно. Прежде чем объединять наборы данных, присвойте им разные имена с помощью Dataset.as и укажите столбец, используя полное имя, например:

df.as("a").join(df.as("b"), $"a.id" > $"b.id").

Вы также можете установить для spark.sql.analyzer.failAmbigousSelfJoin значение false, чтобы отключить эту проверку.

Это мой код:
df_X = df1.filter(df1['_change_type'] == 'update_preimage') df_Y = df1.filter(df1['_change_type'] == 'update_postimage') dfX.show() dfY.show() из pyspark.sql.functions импорт столбца, массива, освещения, когда, array_remove # получаем условия для всех столбцов, кроме id # условия_ = [when(dfX[c]!=dfY[c],lit(c)).otherwise("") for c в dfX.columns if c != ['external_id', '_change_type']] select_expr =[ col("external_id"), *[dfY[c] для c в dfY.columns if c != 'external_id'], # array_remove(array(*conditions_), "").alias("column_names") ] печать (select_expr) dfX.join(dfY, "external_id").select(*select_expr).show()```