мне надо доработать код который берет два файла, читает через пандас и сравнивает все поля входа и выхода через спарк. надо переделать не используя юнион
import pandas as pd
from pyspark.sql import SparkSession
from pyspark.sql.functions import col, from_json, lit
from pyspark.sql.types import MapType, StringType

# --- 1. Чтение CSV через pandas ---
lt_path = "credit_pot_LT_data20250703.csv"       # замените на свои пути
prom_path = "credit_pot_prom_data20250630.csv"

lt_pdf = pd.read_csv(lt_path, dtype=str)
prom_pdf = pd.read_csv(prom_path, dtype=str)

from pyspark.sql import SparkSession
from pyspark.sql.functions import col, from_json, lit, udf, array_distinct, flatten, map_keys, union, array, expr
from pyspark.sql.types import MapType, StringType, ArrayType

# --- Предположим, lt_df и prom_df уже загружены и подготовлены без underscore в колонках ---

# 1. Схема для JSON
json_schema = MapType(StringType(), StringType())

# 2. Парсим JSON из строк в MapType (если ещё не сделали)
for df,cols in [(lt_df, ['inparams','outparams']), (prom_df,['inparams','outparams'])]:
    for c in cols:
        if c in df.columns:
            df = df.withColumn(c+'_map', from_json(col(c), json_schema)).drop(c)

# Обновляем ссылки, если изменились
# предположим снаружи заменили переменные lt_df и prom_df

# 3. Функция для вычленения всех ключей MapType по всей таблице для заданной колонки
def get_all_map_keys(df, map_col):
    keys_df = df.select(flatten(array_distinct(array([map_keys(col(map_col))]))).alias("keys"))
    # Собираем ключи в список
    keys = keys_df.selectExpr("explode(keys) as key").distinct().collect()
    return sorted([r.key for r in keys if r.key is not None])

# 4. Получаем все ключи из inparams_map и outparams_map по обеим таблицам
lt_in_keys = get_all_map_keys(lt_df, "inparams_map") if "inparams_map" in lt_df.columns else []
lt_out_keys = get_all_map_keys(lt_df, "outparams_map") if "outparams_map" in lt_df.columns else []
prom_in_keys = get_all_map_keys(prom_df, "inparams_map") if "inparams_map" in prom_df.columns else []
prom_out_keys = get_all_map_keys(prom_df, "outparams_map") if "outparams_map" in prom_df.columns else []

all_in_keys = sorted(set(lt_in_keys) | set(prom_in_keys))
all_out_keys = sorted(set(lt_out_keys) | set(prom_out_keys))

# 5. Функция для добавления колонок по ключам из MapType
def expand_map_columns(df, map_col, keys, prefix):
    for k in keys:
        # Извлекаем строковое значение по ключу
        df = df.withColumn(f"{prefix}_{k}", col(map_col).getItem(k))
    return df

lt_df = expand_map_columns(lt_df, "inparams_map", all_in_keys, "inparams")
lt_df = expand_map_columns(lt_df, "outparams_map", all_out_keys, "outparams")

prom_df = expand_map_columns(prom_df, "inparams_map", all_in_keys, "inparams")
prom_df = expand_map_columns(prom_df, "outparams_map", all_out_keys, "outparams")

# 6. Теперь можно удалить map-колонки, они не нужны
lt_df = lt_df.drop("inparams_map", "outparams_map")
prom_df = prom_df.drop("inparams_map", "outparams_map")

# 7. Заполняем NULL в id
lt_df = lt_df.fillna({'id': 'NULL'})
prom_df = prom_df.fillna({'id': 'NULL'})

# 8. Назовем колонки, которые есть в обоих датафреймах, для сравнения
common_cols = list(set(lt_df.columns).intersection(set(prom_df.columns)))
common_cols.remove("id")  # id отдельно

# 9. Join по id
joined = prom_df.alias("p").join(lt_df.alias("l"), on="id", how="inner")

from pyspark.sql.functions import when

# 10. Сравним колонки
compare_exprs = []
for c in common_cols:
    compare_exprs.append((col(f"p.{c}") == col(f"l.{c}")).alias(f"{c}_diff"))

# 11. Итоговые колонки (id + пары + diff)
result_cols = [col("p.id").alias("id")]

for c in common_cols:
    result_cols.append(col(f"p.{c}").alias(f"{c}_prom"))
for c in common_cols:
    result_cols.append(col(f"l.{c}").alias(f"{c}_lt"))

result_cols.extend(compare_exprs)

result_df = joined.select(*result_cols)

result_df.show(truncate=False)

ImportError: cannot import name 'union'