у меня есть файл из которого собирается первая таблица (код ниже, полностью рабочий для парсинга, не меняй его), а также таблица в КП, из которой надо подтянуть данные для обработки и сравнения с первой таблицей. код ниже не учитывает, что мне нужны все колонки из первой таблицы в результирующем дф и вот эти колонки второй таблицы, которые нужны inn1 (string) epk_id1 (string) inn2 (string) epk_id2 (string) crit (string)
вторая таблица содержит данные по участникам в разрезе участников за которые отвечает epk_id2, заемщики в epk_id1.
проанализируй код и пойми суть с учетом добавления данных в результат. для понимания суть в том что надо взять таблицу из файла распарсенного и сгруппировать по заемщикам (epk заемщика), заемщики имеют роль и роль детайлд 1, а участники роль 29, остальных отсечь. то есть группировка заемщик и участники через ;
тоже самое надо сделать с таблицей из КП, но там уже нет ролей, поэтому просто сгруппировать по заемщикам, которые представлены в колонке epk_id1, другие колонки должны содержать через точку с запятой относящиеся к заемщику данные все. далее надо левым джоином присоединить вторую сгруппированную таблицу к первой сгруппированной по epk_id заемщика. далее надо сделать колонки различий не только по epk. надо также сделать колонку с критериями (crit из второй таблицы). как список полный критериев, так и отдельной колонкой критерии по тем epk_id которые нашлись во второй таблице но не нашлись в первой. то есть каждый критерий должен один к одному соотноситься с epk_id участника (исходя из структуры второй таблицы)
ТО ЕСТЬ В ИТОГЕ ДОЛЖНО ПОЛУЧИТЬСЯ СТОЛЬКО СТРОК, СКОЛЬКО ЕПК ЗАЕМЩИКА В ПЕРВОЙ СГРУППИРОВАННОЙ ТАБЛИЦЕ, ПОСКОЛЬКУ МЫ ГРУППИРУЕМ ПЕРВУЮ ТАБЛИЦУ ПО ЕПК ЗАЕМЩИКА (РОЛЬ И РОЛЬ ДЕТАЙЛД 1) И ПРИСОЕДИНЯЕМ ЛЕВЫМ ДЖОИНОМ ВТОРУЮ СГРУППИРОВАННУЮ ПО epk_id1 ТАБЛИЦУ
import pandas as pd
import numpy as np
import xml.etree.ElementTree as ET
from tqdm import tqdm

from pyspark.sql import SparkSession
from pyspark.sql.functions import col, collect_set, array_except, array, array_join, udf
from pyspark.sql.types import StringType, ArrayType

Пути к csv
filename_promJ = 'DSME2506.csv' # файл с XML колонкой request_data_modern

Парсим XML из csv, формируем колоноки:
rquid, дата, inn заемщика, role, role_detailed, inn, epk_id, кол-во участников
def parse_xml_request_data(xml_str):
try:
root = ET.fromstring(xml_str)
except Exception as e:
Если парсинг сломался, возвращаем None
return None

rquid = root.findtext('header/rqUid')
date = root.findtext('header/rqTm')
participants = root.findall('request/participants/participant')

result_rows = []

Сохраним inn заемщика для дальней использования
inn_zayemshik = None

for participant in participants:

Попытка взять INN, в порядке naturalPerson/inn, legalPerson/inn, organizatization/inn
inn_elem = participant.find('naturalPerson/inn')
if inn_elem is None:
inn_elem = participant.find('legalPerson/inn')
if inn_elem is None:
inn_elem = participant.find('organization/inn')
inn = inn_elem.text if inn_elem is not None else None

role_elem = participant.find('rolesList/roleItem/role')
role_detailed_elem = participant.find('rolesList/roleItem/roleDetailed')
epk_id_elem = participant.find('ids/epkId')

role = role_elem.text if role_elem is not None else None
role_detailed = role_detailed_elem.text if role_detailed_elem is not None else None
epk_id = epk_id_elem.text if epk_id_elem is not None else None

Для role=1 и role_detailed=1 берём inn заемщика (для удобства)
if role == '1' and role_detailed == '1':
inn_zayemshik = inn

result_rows.append({
'rquid': rquid,
'date': date,
'inn_заемщика': inn_zayemshik,
'role': role,
'role_detailed': role_detailed,
'inn': inn,
'epk_id': epk_id,
'кол-во участников': len(participants)
})
return result_rows
Читаем исходный csv
df = pd.read_csv(filename_promJ, dtype=str)

Коллекция для всех разбитых строк
all_rows = []

for x in tqdm(df['request_data_modern'].to_list(), desc='Parsing XML'):
parsed = parse_xml_request_data(x)
if parsed:
all_rows.extend(parsed)

df_parsed = pd.DataFrame(all_rows)

Преобразуем NaN для чтения совместимости
df_parsed.replace({None: np.nan}, inplace=True)

Для следующего шага удобнее положить df_parsed в Spark
spark_df_parsed = spark.createDataFrame(df_parsed.fillna('NULL'))

--- Далее по пунктам ---
2. По строкам с role=1 (заёмщик), для каждого заёмщика формируем список epk_id участников GSЛ с role=29,
где epk_id заполнен (отбрасываем дублеров - у участников эпк_id не должно быть пустым)
Фильтрация заемщиков — role=1, role_detailed=1 (сохраним epk_id заемщика)
borrowers_df = spark_df_parsed.filter((col('role') == '1') & (col('role_detailed') == '1') & (col('epk_id').isNotNull()))

Фильтрация участников ГСЛ — role=29, epk_id не пустой, и тот же rquid (чтобы привязать к заёмщику из этого же запроса)
participants_gsl_df = spark_df_parsed.filter((col('role') == '29') & (col('epk_id').isNotNull()))

Для каждого rquid + epk_id заемщика собираем список epk_id участников с ролью 29
joined = borrowers_df.alias('b').join(
participants_gsl_df.alias('p'),
on=['rquid', 'date', 'inn_заемщика'],
how='left'
).select(
col('b.rquid'),
col('b.date'),
col('b.epk_id').alias('epk_id_заемщика'),
col('p.epk_id').alias('epk_id_участника')
)

Группируем по заёмщику и собираем участников в список (удаляем None)
from pyspark.sql.functions import collect_set, array_remove

grouped = joined.groupby(
'rquid', 'date', 'epk_id_заемщика'
).agg(
collect_set('epk_id_участника').alias('epk_ids_участников')
)

Очистим None из списка if есть
clean_none_udf = udf(lambda lst: [x for x in lst if x is not None], ArrayType(StringType()))
grouped = grouped.withColumn('epk_ids_участников', clean_none_udf(col('epk_ids_участников')))

Для удобного вывода превратим список в строку через запятую
grouped = grouped.withColumn('epk_ids_участников_str', array_join('epk_ids_участников', ';'))

По таблице ГСЛ (t_team_rkib_mle.gsl_full_basis_20250618) отобрать записи, где epk_id1 == epk_id заемщика,
от этих записей сформировать список epk_id2 — положить в колонку 3
Читаем таблицу ГСЛ
gsl_table = spark.table('t_team_rkib_mle.gsl_full_basis_20250618').select(
'epk_id1', 'epk_id2'
).distinct()
Соединяем с epk_id заемщика из grouped, берем epk_id2 по epk_id1 (заемщик)
gsl_filtered = grouped.join(gsl_table, grouped.epk_id_заемщика == gsl_table.epk_id1, how='left')
.select(
grouped['*'],
gsl_table.epk_id2.alias('epk_id2')
)

Группируем для формирования списка epk_id2 по эпк_id заемщика
gsl_grouped = gsl_filtered.groupby(
'rquid', 'date', 'epk_id_заемщика', 'epk_ids_участников', 'epk_ids_участников_str'
).agg(
collect_set('epk_id2').alias('epk_ids2')
)

Очистим None
gsl_grouped = gsl_grouped.withColumn('epk_ids2', clean_none_udf(col('epk_ids2')))
gsl_grouped = gsl_grouped.withColumn('epk_ids2_str', array_join('epk_ids2', ';'))

В колонку 4 положить epk_id участников из пункта 2, которых нет среди epk_id из пункта 3,
а в колонку 5 положить те epk_id которые есть в пункте 3, но нет в пункте 2
Для этого вычислим множество разницы во внешних и внутренних данных
def list_difference(list1, list2):
set1 = set(list1) if list1 else set()
set2 = set(list2) if list2 else set()
return list(set1 - set2)
def list_difference_str(list1, list2):
dif = list_difference(list1, list2)
return ';'.join(dif) if dif else None

diff_udf = udf(list_difference_str, StringType())

Колонка 4: участники epk_id из пункта 2, которых нет в epk_id2 из пункта 3
Колонка 5: наоборот — epk_id2 которые нет в списке участников из пункта 2
final_df = gsl_grouped
.withColumn('участники_нет_в_гсл', diff_udf(col('epk_ids_участников'), col('epk_ids2')))
.withColumn('гсл_нет_в_участниках', diff_udf(col('epk_ids2'), col('epk_ids_участников')))

Для финала сформируем нужные колонки по формату (выведим как pandas для окончательного сохранения csv)
final_pdf = final_df.select(
'rquid',
'date',
'epk_id_заемщика',
'epk_ids_участников_str',
'epk_ids2_str',
'участники_нет_в_гсл',
'гсл_нет_в_участниках'
).toPandas()

Загружаем в Spark DataFrame
spark_df = spark.createDataFrame(df_parsed.fillna('NULL'))

Фильтруем роли заемщика (1) и участников (29), удаляя остальные роли и строки с 'NULL' epk_id
filtered_df = spark_df.filter(
((col('role') == '1') | (col('role') == '29')) & (col('epk_id') != 'NULL')
)

Для заемщиков (role = 1) берем epk_id, inn заемщика и кол-во участников (берем distinct)
borrowers_df = filtered_df.filter(col('role') == '1').select(
col('epk_id').alias('epk_id_заемщика'),
col('inn_заемщика'),
col('кол-во участников')
).distinct()

Для участников (role = 29) берем rquid, epk_id участника
participants_df = filtered_df.filter(col('role') == '29').select(
'rquid',
col('epk_id').alias('epk_id_участника')
).distinct()

Получаем связь rquid -> epk_id_заемщика для заемщиков
borrowers_rquid = filtered_df.filter(col('role') == '1').select('rquid', col('epk_id').alias('epk_id_заемщика')).distinct()

Присоединяем участников к заемщикам по rquid
joined = borrowers_rquid.join(
participants_df,
on='rquid',
how='left'
).select(
'epk_id_заемщика',
'epk_id_участника'
)

Группируем участников по epk_id_заемщика, собираем уникальные epk_id участников
grouped_participants = joined.groupBy('epk_id_заемщика').agg(
collect_set('epk_id_участника').alias('список_epk_участников')
)

Объединяем заемщиков с их списком участников
borrowers_with_participants = borrowers_df.join(
grouped_participants,
on='epk_id_заемщика',
how='left'
)

Заменяем null в списках участников на пустой массив
borrowers_with_participants = borrowers_with_participants.withColumn(
'список_epk_участников',
when(col('список_epk_участников').isNull(), array()).otherwise(col('список_epk_участников'))
)

Получаем таблицу ГСЛ из Spark
gsl_table = spark.table('t_team_rkib_mle.gsl_full_basis_20250618').select(
'epk_id1',
'epk_id2'
).distinct()

Группируем таблицу ГСЛ по epk_id1, собираем список epk_id2
gsl_grouped = gsl_table.groupBy('epk_id1').agg(
collect_set('epk_id2').alias('гсл_epk_ids2')
)

Левый join по epk_id заемщика
final_df = borrowers_with_participants.join(
gsl_grouped.withColumnRenamed('epk_id1', 'epk_id_заемщика'),
on='epk_id_заемщика',
how='left'
)

Заменяем null в gsl списках на пустой массив
final_df = final_df.withColumn(
'гсл_epk_ids2',
when(col('гсл_epk_ids2').isNull(), array()).otherwise(col('гсл_epk_ids2'))
)

UDF для вычисления разницы множеств, возвращает строку через ;
def list_diff_str(a, b):
set_a = set(a) if a else set()
set_b = set(b) if b else set()
diff = set_a - set_b
return ';'.join(diff) if diff else None

diff_udf = udf(list_diff_str, StringType())

Добавляем колонки с разницами множеств
final_df = final_df
.withColumn('участники_нет_в_гсл', diff_udf(col('список_epk_участников'), col('гсл_epk_ids2')))
.withColumn('гсл_нет_в_участниках', diff_udf(col('гсл_epk_ids2'), col('список_epk_участников')))

Дополнительно переведем списки в строку с разделителем для удобного экспорта
final_df = final_df
.withColumn('список_epk_участников_str', array_join('список_epk_участников', ';'))
.withColumn('гсл_epk_ids2_str', array_join('гсл_epk_ids2', ';'))