суть в том что у меня есть два датафрейма (пром и тест) с входными и исходящими данными в формате json. мне надо распарсить максимально их и сравнить по каждому параметру. сохранить все параметры с дополнительными колонками наличия расхождений по ним (diff). и в конце добавить результирующие колонки по расхождениям входных и исходящих параметров. у меня получился такой код, но он несогласован и не до конца оптимизирован. исправь его. напиши полный код
from pyspark.sql.functions import col, when, lit
import json
from pyspark.sql.functions import udf, col, when, lit, array, size, sort_array, concat_ws
from pyspark.sql.types import StringType, MapType

def flatten_json(data, prefix=''):
    """
    Рекурсивно "сплющивает" вложенный объект JSON в dict ключ→значение, где ключ — путь с точками и индексами.
    Пример: {'a': {'b': [1,2]}} → {'a.b[0]': '1', 'a.b[1]': '2'}
    Все значения приводим к строке.
    """
    flat = dict()
    if isinstance(data, dict):
        for k, v in data.items():
            new_key = f"{prefix}.{k}" if prefix else k
            flat.update(flatten_json(v, new_key))
    elif isinstance(data, list):
        for i, v in enumerate(data):
            new_key = f"{prefix}[{i}]"
            flat.update(flatten_json(v, new_key))
    else:
        # Leaf node
        flat[prefix] = 'NULL' if data is None else str(data)
    return flat

flatten_udf = udf(lambda x: flatten_json(json.loads(x) if x else {}), MapType(StringType(), StringType()))


# Загружаем
df_prom = spark.table("t_team_mmb_od.pprb_rr3_dream_test")
df_test = spark.table("t_team_mmb_od.pprb_rr3_dream_test_out")

# Добавляем колонку map_inparams с флэттенингом JSON из inparams, то же для outparams
df_prom_flat = df_prom \
    .withColumn("map_inparams", flatten_udf(col("inparams"))) \
    .withColumn("map_outparams", flatten_udf(col("outparams"))) \
    .select("chainid", "map_inparams", "map_outparams")

df_test_flat = df_test \
    .withColumn("map_inparams", flatten_udf(col("inparams"))) \
    .withColumn("map_outparams", flatten_udf(col("outparams"))) \
    .select("chainid", "map_inparams", "map_outparams")


joined = df_prom_flat.alias("prom").join(
    df_test_flat.alias("test"),
    on="chainid", how="inner"
)

from pyspark.sql.functions import map_keys, array_union

keys_inparams = joined.select(
    array_union(
        map_keys(col("prom.map_inparams")),
        map_keys(col("test.map_inparams"))
    ).alias("keys_inparams"),
    array_union(
        map_keys(col("prom.map_outparams")),
        map_keys(col("test.map_outparams"))
    ).alias("keys_outparams")
).limit(1).collect()[0]

keys_inparams_list = keys_inparams["keys_inparams"]
keys_outparams_list = keys_inparams["keys_outparams"]

new_columns = []

# Исходные колонки
base_cols = [col("chainid")]

# Добавляем существующие поля
new_columns.extend(base_cols)

# Генерируем новые колонки
for key in tqdm(keys_inparams_list):
    prom_val = col("prom.map_inparams").getItem(key)
    test_val = col("test.map_inparams").getItem(key)

    new_columns.extend([
        when(prom_val.isNull(), lit('NULL')).otherwise(prom_val).alias(f"inparams_{key}_prom"),
        when(test_val.isNull(), lit('NULL')).otherwise(test_val).alias(f"inparams_{key}_test"),
        when(
            (prom_val.isNull() & test_val.isNull()) | (prom_val == test_val),
            lit(False)
        ).otherwise(lit(True)).alias(f"inparams_{key}_diff")
    ])

# То же для outparams...
for key in tqdm(keys_outparams_list):
    prom_val = col("prom.map_outparams").getItem(key)
    test_val = col("test.map_outparams").getItem(key)

    new_columns.extend([
        when(prom_val.isNull(), lit('NULL')).otherwise(prom_val).alias(f"outparams_{key}_prom"),
        when(test_val.isNull(), lit('NULL')).otherwise(test_val).alias(f"outparams_{key}_test"),
        when(
            (prom_val.isNull() & test_val.isNull()) | (prom_val == test_val),
            lit(False)
        ).otherwise(lit(True)).alias(f"outparams_{key}_diff")
    ])
# Финальный select
joined = joined.select(*new_columns)

compare_udf = udf(
    lambda prom_dict, test_dict: ", ".join(
        f"{key}: {prom_dict[key]}::{test_dict.get(key, 'MISSING')}"
        for key in sorted(set(prom_dict.keys()) | set(test_dict.keys()))
        if prom_dict.get(key) != test_dict.get(key)
    ) if prom_dict or test_dict else "",
    StringType()
)

final_df = joined.withColumn(
    "inparams_diff",
    compare_udf(col("prom.map_inparams"), col("test.map_inparams"))
).withColumn(
    "outparams_diff",
    compare_udf(col("prom.map_outparams"), col("test.map_outparams"))
).select("chainid", "inparams_diff", "outparams_diff")