Как переместить SQL-код в PySpark, используя агрегаты и вложенные запросы?

Question

Я пытаюсь переместить следующий код в PySpark. Не могли бы вы помочь мне? Я новичок в PySpark и должен спешить с этим.

Нет необходимости создавать кадры данных в коде, как я уже эти таблицы создал. Основная проблема, с которой я сталкиваюсь, заключается в создании столбцов суммы уже с другими атрибутами в той же таблице (2-й и 3-й запросы), но я не могу преобразовать ни один из них.

SELECT  attr_data_provider_key,
    CUST_KEY,
    prod_key,
    category_name,
    cust_name,
    time_period_end_date,
    SUM(SU) AS SU
FROM KPI_BOP_STEP1
WHERE attr_data_provider_key IN (SELECT attr_data_provider_key 
                                    FROM KPI_BOP_TARGET
                             WHERE CUSTOMER_TEAM_BOP_LVL = "CUSTOMER"
                             AND PRODUCT_BOP_LVL = "CATEGORY")
GROUP BY        attr_data_provider_key,
            CUST_KEY,
            prod_key,
            category_name,
            time_period_end_date,
            cust_name;

SELECT  BOP1.time_period_end_date,
    BOP1.attr_data_provider_key,
    BOP1.CUST_KEY,
    BOP1.prod_key,
    BOP1.category_name,
    BOP1.cust_name,
    BOP1.su/BOP2.SUM_su AS SHARE
FROM    KPI_BOP_STEP2 BOP1
    LEFT JOIN (
                    SELECT  time_period_end_date,
                            attr_data_provider_key,
                            cust_name,
                            category_name,
                            SUM(SU) SUM_SU
                    FROM    KPI_BOP_STEP2
                    GROUP BY attr_data_provider_key,
                             cust_name,
                             category_name,
                             time_period_end_date
                     ) BOP2 
    ON (    BOP1.attr_data_provider_key = BOP2.attr_data_provider_key 
            AND BOP1.cust_name = BOP2.cust_name
            AND BOP1.category_name = BOP2.category_name
            and BOP1.time_period_end_date = BOP2.time_period_end_date);

SELECT  time_period_end_date,
    CUST_KEY,
    prod_key,
    SHARE * BOP_VOLUME_MSU AS TARGET
FROM KPI_BOP_CUSTCAT KPI
JOIN KPI_BOP_TARGET BOP ON (BOP.attr_data_provider_key = 
KPI.attr_data_provider_key
                            AND BOP.CUSTOMER_TEAM = KPI.cust_name
                            AND BOP.PRODUCT = KPI.category_name
                            AND last_day(BOP.BOP_MONTH) = KPI.time_period_end_date);