помоги, у меня код который берет два дф и в спарке распаршивает и сравнивает все параметры из json строк для колонок
from pyspark.sql.functions import col, from_json, explode, udf, array_sort, split, array, lit, when, concat_ws
from pyspark.sql.types import StructType, StringType, MapType
import json
import re

# --- 1. Загружаем исходные датафреймы ---
lt_path = 'credit_pot_LT_data20250703.csv'  # или parquet
prom_path = 'credit_pot_prom_data20250630.csv'
lt_pdf = pd.read_csv(lt_path)
prom_pdf = pd.read_csv(prom_path)
lt_df = spark.createDataFrame(lt_pdf)  # адаптируйте к типу файла
prom_df = spark.createDataFrame(prom_pdf)

# --- 2. Колонка 'id' индексируем, заполним NULL для сравнения ---
lt_df = lt_df.fillna('NULL', subset=["id"]).withColumnRenamed("id", "id")
prom_df = prom_df.fillna('NULL', subset=["id"]).withColumnRenamed("id", "id")

# --- 3. Находим колонки с 'inparams' и 'outparams' (в любом варианте) ---
def find_json_cols(columns):
    keywords = ['inparams', 'outparams', 'in_params', 'out_params']
    return [col_name for col_name in columns if any(kw in col_name.lower() for kw in keywords)]

lt_json_cols = find_json_cols(lt_df.columns)
prom_json_cols = find_json_cols(prom_df.columns)

# Общий набор колонок для парсинга (можно сделать объединение или отдельно)
all_json_cols = list(set(lt_json_cols).union(set(prom_json_cols)))

# --- 4. Функция для инферирования схемы JSON ---
def infer_json_schema(df, col_name):
    # Пример: берём несколько строк, пробуем парсить и строим MapType()
    # Тут можно более сложно через sampling и spark.read.json, но для простоты используем MapType()
    return MapType(StringType(), StringType())

schema_map = {}
for c in all_json_cols:
    schema_map[c] = infer_json_schema(lt_df.union(prom_df), c)

# --- 5. Распарсим JSON колонки ---
def expand_json_columns(df, json_columns):
    for c in json_columns:
        # Переименование в префикс для колонки
        prefix = c.replace('in_params', 'inparams').replace('out_params', 'outparams')
        # Парсим json из строки в MapType
        df = df.withColumn(c + '_map', from_json(col(c), schema_map[c]))
        # Генерируем колонки из мапы
        df = df.select(
            "*",
            *[col(c + '_map').getItem(k).alias(f"{prefix}_{k}") for k in df.select(c + '_map').schema[0].dataType.keyType._metadata.keys()]  # Вариант для ключей динамически
        )
        df = df.drop(c).drop(c + '_map')
    return df

# Однако, ключи из MapType не всегда знают. Для универсальности надо:
# - собирать ключи из всех строк
# - или парсить вручную

# Для простоты и скорости можно сделать udf для парсинга и преобразования словаря в колонки при необходимости, или оставить как MapType.

# Альтернативный вариант — оставить MapType или StructType и сравнивать их, но с доступом к ключам.

# Поэтому предлагаем упростить: оставить MapType столбцы с json и сравнивать целиком.

for c in all_json_cols:
    schema = MapType(StringType(), StringType())
   prom_df = prom_df.withColumn(c + '_map', from_json(col(c), schema)).drop(c)
    lt_df = lt_df.withColumn(c + '_map', from_json(col(c), schema)).drop(c)

# --- 6. Соединяем датафреймы по id ---
joined_df = prom_df.alias('p').join(lt_df.alias('l'), on='id', how='inner')

# --- 7. Сравним соответствующие колонки ---
# Для колонок без JSON осталось:
plain_cols = [c for c in prom_df.columns if c not in [*map(lambda x: x + '_map', all_json_cols), 'id']]
compare_exprs = []

for c in plain_cols:
    compare_exprs.append((col(f'p.{c}') == col(f'l.{c}')).alias(c + '_diff'))

for c in all_json_cols:
    # Сравниваем map колонку целиком:
    compare_exprs.append((col(f'p.{c}_map') == col(f'l.{c}_map')).alias(c + '_diff'))

# --- 8. Работа с кодами (полями, содержащими 'code') ---
# Допустим, колонки code_prom и code_test (их надо переименовать в prom и test внутри joined_df)
# В вашем примере именование из prom и l (lt_df)

# Если есть колонки с названием 'code' в prom и lt_df (например 'code'):
# Добавим эти колонки в joined_df, если есть
if 'code' in prom_df.columns and 'code' in lt_df.columns:
    def code_process(column):
        # Функция для сортировки частей кода с числами
        def sorter_func(codestr):
            parts = codestr.split(";") if codestr else []
            try:
                sorted_parts = sorted(parts, key=lambda x: int(re.search(r'\d+', x).group()) if re.search(r'\d+', x) else float('inf'))
                return ";".join(sorted_parts)
            except:
                return codestr
        return udf(sorter_func, StringType())(column)

    joined_df = joined_df.withColumn("code_prom", col("p.code"))
    joined_df = joined_df.withColumn("code_test", col("l.code"))

    joined_df = joined_df.fillna({'code_prom': 'NULL', 'code_test': 'NULL'}) \
        .withColumn("code_prom_sorted", code_process(col("code_prom"))) \
        .withColumn("code_test_sorted", code_process(col("code_test")))

    # Сравнение как множества
    def codes_equal(prom_str, test_str):
        set_prom = set(prom_str.split(";")) if prom_str else set()
        set_test = set(test_str.split(";")) if test_str else set()
        return set_prom == set_test

    codes_equal_udf = udf(codes_equal, StringType())

    joined_df = joined_df.withColumn("code_diff", (codes_equal_udf(col("code_prom_sorted"), col("code_test_sorted")) == "true"))

else:
    joined_df = joined_df.withColumn("code_diff", lit(None))

# --- 9. Итоговые колонки ---
result_cols = ['id'] + plain_cols + \
              [c + '_diff' for c in plain_cols] + \
              [c + '_diff' for c in all_json_cols] + \
              ['code_prom', 'code_test', 'code_diff']

result_df = joined_df.select(*[col("p.id").alias("id")] + \
                             [col(f'p.{c}').alias(f"{c}_prom") for c in plain_cols] + \
                             [col(f'l.{c}').alias(f"{c}_test") for c in plain_cols] + \
                             [col(f"{c}_diff") for c in [*plain_cols, *all_json_cols]] + \
                             ['code_prom', 'code_test', 'code_diff'])

# --- 10. Сохраняем или выводим результат ---
result_df.show(truncate=False)
# Или result_df.write.csv(...), parquet и т.п.
# .repartition(60).write.saveAsTable('t_team_mmb_od.rr3_regress_test')

---------------------------------------------------------------------------
Py4JJavaError                             Traceback (most recent call last)
/usr/sdp/current/spark2-client/python/pyspark/sql/utils.py in deco(*a, **kw)
     62         try:
---> 63             return f(*a, **kw)
     64         except py4j.protocol.Py4JJavaError as e:

/usr/sdp/current/spark2-client/python/lib/py4j-0.10.7-src.zip/py4j/protocol.py in get_return_value(answer, gateway_client, target_id, name)
    327                     "An error occurred while calling {0}{1}{2}.\n".
--> 328                     format(target_id, ".", name), value)
    329             else:

Py4JJavaError: An error occurred while calling o388.union.
: org.apache.spark.sql.AnalysisException: Union can only be performed on tables with the same number of columns, but the first table has 6 columns and the second table has 13 columns;;
'Union
:- Project [chain_id#0, in_params#1, out_params#2, id#3L AS id#50L, date_exec#4, rw#5L]
:  +- Project [chain_id#0, in_params#1, out_params#2, id#3L, date_exec#4, rw#5L]
:     +- LogicalRDD [chain_id#0, in_params#1, out_params#2, id#3L, date_exec#4, rw#5L], false
+- Project [chainid#12, strategyid#13, dateexec#14, inparams#15, outparams#16, getMmbScore#17, getIrbSmeSprV2Score#18, Collect_data#19, Collect_data3#20, id#21L AS id#83L, exec_time#22, model_code#23, Rw#24]
   +- Project [chainid#12, strategyid#13, dateexec#14, inparams#15, outparams#16, getMmbScore#17, getIrbSmeSprV2Score#18, Collect_data#19, Collect_data3#20, id#21L, exec_time#22, model_code#23, Rw#24]
      +- LogicalRDD [chainid#12, strategyid#13, dateexec#14, inparams#15, outparams#16, getMmbScore#17, getIrbSmeSprV2Score#18, Collect_data#19, Collect_data3#20, id#21L, exec_time#22, model_code#23, Rw#24], false

                at org.apache.spark.sql.catalyst.analysis.CheckAnalysis$class.failAnalysis(CheckAnalysis.scala:43)
                at org.apache.spark.sql.catalyst.analysis.Analyzer.failAnalysis(Analyzer.scala:95)
                at org.apache.spark.sql.catalyst.analysis.CheckAnalysis$$anonfun$checkAnalysis$1$$anonfun$apply$12.apply(CheckAnalysis.scala:283)
                at org.apache.spark.sql.catalyst.analysis.CheckAnalysis$$anonfun$checkAnalysis$1$$anonfun$apply$12.apply(CheckAnalysis.scala:280)
                at scala.collection.immutable.List.foreach(List.scala:392)
                at org.apache.spark.sql.catalyst.analysis.CheckAnalysis$$anonfun$checkAnalysis$1.apply(CheckAnalysis.scala:280)
                at org.apache.spark.sql.catalyst.analysis.CheckAnalysis$$anonfun$checkAnalysis$1.apply(CheckAnalysis.scala:86)
                at org.apache.spark.sql.catalyst.trees.TreeNode.foreachUp(TreeNode.scala:126)
                at org.apache.spark.sql.catalyst.analysis.CheckAnalysis$class.checkAnalysis(CheckAnalysis.scala:86)
                at org.apache.spark.sql.catalyst.analysis.Analyzer.checkAnalysis(Analyzer.scala:95)
                at org.apache.spark.sql.catalyst.analysis.Analyzer$$anonfun$executeAndCheck$1.apply(Analyzer.scala:108)
                at org.apache.spark.sql.catalyst.analysis.Analyzer$$anonfun$executeAndCheck$1.apply(Analyzer.scala:105)
                at org.apache.spark.sql.catalyst.plans.logical.AnalysisHelper$.markInAnalyzer(AnalysisHelper.scala:201)
                at org.apache.spark.sql.catalyst.analysis.Analyzer.executeAndCheck(Analyzer.scala:105)
                at org.apache.spark.sql.execution.QueryExecution.analyzed$lzycompute(QueryExecution.scala:58)
                at org.apache.spark.sql.execution.QueryExecution.analyzed(QueryExecution.scala:56)
                at org.apache.spark.sql.execution.QueryExecution.assertAnalyzed(QueryExecution.scala:48)
                at org.apache.spark.sql.Dataset$.ofRows(Dataset.scala:78)
                at org.apache.spark.sql.Dataset.withSetOperator(Dataset.scala:3424)
                at org.apache.spark.sql.Dataset.union(Dataset.scala:1862)
                at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
                at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
                at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
                at java.lang.reflect.Method.invoke(Method.java:498)
                at py4j.reflection.MethodInvoker.invoke(MethodInvoker.java:244)
                at py4j.reflection.ReflectionEngine.invoke(ReflectionEngine.java:357)
                at py4j.Gateway.invoke(Gateway.java:282)
                at py4j.commands.AbstractCommand.invokeMethod(AbstractCommand.java:132)
                at py4j.commands.CallCommand.execute(CallCommand.java:79)
                at py4j.GatewayConnection.run(GatewayConnection.java:238)
                at java.lang.Thread.run(Thread.java:750)


During handling of the above exception, another exception occurred:

AnalysisException                         Traceback (most recent call last)
<ipython-input-4-4e24c51fc95f> in <module>
     35 schema_map = {}
     36 for c in all_json_cols:
---> 37     schema_map[c] = infer_json_schema(lt_df.union(prom_df), c)
     38 
     39 # --- 5. Распарсим JSON колонки ---

/usr/sdp/current/spark2-client/python/pyspark/sql/dataframe.py in union(self, other)
   1459         Also as standard in SQL, this function resolves columns by position (not by name).
   1460         """
-> 1461         return DataFrame(self._jdf.union(other._jdf), self.sql_ctx)
   1462 
   1463     @since(1.3)

/usr/sdp/current/spark2-client/python/lib/py4j-0.10.7-src.zip/py4j/java_gateway.py in __call__(self, *args)
   1255         answer = self.gateway_client.send_command(command)
   1256         return_value = get_return_value(
-> 1257             answer, self.gateway_client, self.target_id, self.name)
   1258 
   1259         for temp_arg in temp_args:

/usr/sdp/current/spark2-client/python/pyspark/sql/utils.py in deco(*a, **kw)
     67                                              e.java_exception.getStackTrace()))
     68             if s.startswith('org.apache.spark.sql.AnalysisException: '):
---> 69                 raise AnalysisException(s.split(': ', 1)[1], stackTrace)
     70             if s.startswith('org.apache.spark.sql.catalyst.analysis'):
     71                 raise AnalysisException(s.split(': ', 1)[1], stackTrace)

AnalysisException: "Union can only be performed on tables with the same number of columns, but the first table has 6 columns and the second table has 13 columns;;\n'Union\n:- Project [chain_id#0, in_params#1, out_params#2, id#3L AS id#50L, date_exec#4, rw#5L]\n:  +- Project [chain_id#0, in_params#1, out_params#2, id#3L, date_exec#4, rw#5L]\n:     +- LogicalRDD [chain_id#0, in_params#1, out_params#2, id#3L, date_exec#4, rw#5L], false\n+- Project [chainid#12, strategyid#13, dateexec#14, inparams#15, outparams#16, getMmbScore#17, getIrbSmeSprV2Score#18, Collect_data#19, Collect_data3#20, id#21L AS id#83L, exec_time#22, model_code#23, Rw#24]\n   +- Project [chainid#12, strategyid#13, dateexec#14, inparams#15, outparams#16, getMmbScore#17, getIrbSmeSprV2Score#18, Collect_data#19, Collect_data3#20, id#21L, exec_time#22, model_code#23, Rw#24]\n      +- LogicalRDD [chainid#12, strategyid#13, dateexec#14, inparams#15, outparams#16, getMmbScore#17, getIrbSmeSprV2Score#18, Collect_data#19, Collect_data3#20, id#21L, exec_time#22, model_code#23, Rw#24], false\n"