при выполнении кода ниже с раскомментаренными строками получаю такую ошибку. если оставить комментирование, все хорошо выполняется. мне нужно чтобы коментированный код работал. то есть смысл в том чтобы взять два датафрейма и сравнить распарсенные json на входе и на выходе каждого дф. при этом если я сравниваю списки (параметры с контейнерами), то мне надо отсортировать их потому что может быть одинаковый результат по содержанию, но разный по порядку, мне важно содержание.
вот ошибка
Py4JJavaError	Traceback (most recent call last)
<ipython-input-7-d57322301fl4> in <module> ).limit(l)
-> 63 keysrow = keysdf collect^ )[0]
keysinparams = keysrow "keysinparams"] keys_outparams keys_row "keysoutparams"]
/usr/sdp/current/spark2-client/python/pyspark/sql/dataframe.py in collect(self)
--> 534
.ith SCCallSiteSync(self _sc) as css
sock_info = self _jdf collectToPython()
list(_load_from_socket(sock_info BatchedSerializer PickleSerializer())))
/usr/sdp/current/spark2-client/python/lib/py4j-0.10.7-src.zip/py4j/java_gateway.py in ___call__(self, *args answer self gateway_client send commandI command) return_value = get_return_value(
-> 1257	answer, self.gateway_client, self.target_id, self.name)
вот код
from pyspark.sql import SparkSession
from pyspark.sql.functions import col, udf, map_keys, array_union, explode, lit, when, coalesce, concat_ws, collect_list, size, array_contains, expr, create_map
from pyspark.sql.types import StringType, MapType
import json
import re

def sanitize_key(k):
   return re.sub(r'[.\[\]]', '_', k)

# --- 1. Функция для "сплющивания" JSON ---
def flatten_json(data, prefix=''):
    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):
        # if isinstance(data[0], dict):
        #     first_key = data[0].keys()[0]
        #     data.sort(key=lambda el: el[first_key])
        for i, v in enumerate(data):
            new_key = f"{prefix}[{i}]"
            flat.update(flatten_json(v, new_key))
    else:
        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()))

# --- 2. Загрузка датафреймов ---
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")

# --- 3. Создание колонок с "сплющенными" JSON ---
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")

# --- 4. Объединяем по chainid ---
joined = df_prom_flat.alias("prom").join(
    df_test_flat.alias("test"),
    on="chainid", how="inner"
)

# --- 5. Получаем все уникальные ключи (для inparams и outparams) один раз ---
keys_df = 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)

keys_row = keys_df.collect()[0]
keys_inparams = keys_row["keys_inparams"]
keys_outparams = keys_row["keys_outparams"]

# --- 6. Функция, создающая колонки для каждой ключа с сравнением ---
def build_param_columns(prefix, keys):
    cols = []
    for key in keys:
        safe_key = sanitize_key(key)
        prom_val = col(f"prom.map_{prefix}").getItem(key)
        test_val = col(f"test.map_{prefix}").getItem(key)

        # Значения с заменой null на 'NULL' (строка)
        prom_val_fix = when(prom_val.isNull(), lit('NULL')).otherwise(prom_val)
        test_val_fix = when(test_val.isNull(), lit('NULL')).otherwise(test_val)

        # Колонка с исходным значением prom
        cols.append(prom_val_fix.alias(f"{prefix}_{safe_key}_prom"))
        # Колонка с исходным значением test
        cols.append(test_val_fix.alias(f"{prefix}_{safe_key}_test"))
        # Колонка с флагом расхождения (True - есть отличие)
        diff_col = when(prom_val_fix == test_val_fix, lit(False)).otherwise(lit(True))
        cols.append(diff_col.alias(f"{prefix}_{safe_key}_diff"))
    return cols

# --- 7. Формируем колонки для inparams и outparams ---
param_columns = []
param_columns += build_param_columns('inparams', keys_inparams)
param_columns += build_param_columns('outparams', keys_outparams)

# В итоге select chainid + все эти колонки
result_df = joined.select(col("chainid"), *param_columns)

# --- 8. Добавляем итоговые колонки с общей проверкой на любые расхождения ---
# Получаем списки колонок-flag (_diff)
in_diff_cols = [c for c in result_df.columns if c.startswith('inparams_') and c.endswith('_diff')]
out_diff_cols = [c for c in result_df.columns if c.startswith('outparams_') and c.endswith('_diff')]

# Суммарный diff флаг (True, если есть хоть одно True среди diff колонок)
result_df = result_df.withColumn("inparams_diff_exists", expr(f"array_contains(array({','.join(in_diff_cols)}), true)")) \
    .withColumn("outparams_diff_exists", expr(f"array_contains(array({','.join(out_diff_cols)}), true)"))

# --- 9. (Опционально) Добавляем итоговые колонки с описанием несоответствий ---
# Можно собрать в одну строку все расхождения по параметрам

def diff_summary_expr(prefix, keys):
    # Формируем массив строк вида key: prom_val::test_val, но только для различающихся
    expressions = []
    for key in keys:
        safe_key = sanitize_key(key)
        prom_col = f"{prefix}_{safe_key}_prom"
        test_col = f"{prefix}_{safe_key}_test"
        diff_col = f"{prefix}_{safe_key}_diff"
        # concat строка с отображением key и значений из prom и test где есть diff
        expr_str = f"CASE WHEN {diff_col} THEN CONCAT('{key}: ', {prom_col}, '::', {test_col}) ELSE NULL END"
        expressions.append(expr_str)
    array_expr = f"filter(array({','.join(expressions)}), x -> x is not null)"
    res_expr = f"concat_ws(', ', {array_expr})"
    return expr(res_expr)

result_df = result_df \
    .withColumn("inparams_diff_details", diff_summary_expr('inparams', keys_inparams)) \
    .withColumn("outparams_diff_details", diff_summary_expr('outparams', keys_outparams))

# --- 10. Итоговые колонки и вывод ---
final_columns = ["chainid"] + \
               [c for c in result_df.columns if c not in ("chainid", "inparams_diff_exists", "outparams_diff_exists", "inparams_diff_details", "outparams_diff_details")] + \
               ["inparams_diff_exists", "outparams_diff_exists", "inparams_diff_details", "outparams_diff_details"]

final_df = result_df.select(*final_columns)