Как эффективно создать отсутствующие образцы дат в наборе данных с помощью PySpark?

Question

Чтобы упростить объяснение вопроса, я создал этот код для создания фиктивного фрейма данных:

import numpy as np
import pandas as pd

np.random.seed(0)

df = pd.DataFrame(data=(
    [
        {'date': date, 'idC': idC, 'idP': idP, 'value': np.random.rand()}
        for date in sorted(np.random.choice(dates, size=nsamples, replace=False))
        for idC in idC_list
        for idP in idP_list
    ]
)).sort_values(['idP', 'idC', 'date']).reset_index(drop=True)

Фальшивый фрейм данных содержит дату, идентификатор ребенка, идентификатор родителя и значение. Затем конечным результатом является кадр данных, в котором недостающие даты в интервале дат создаются со значением NaN.

      date  idC   idP     value
2000-01-02   10  1000  0.437587
2000-01-03   10  1000  0.891773
2000-01-05   10  1000  0.963663
2000-01-07   10  1000  0.383442
2000-01-09   10  1000  0.791725
2000-01-10   10  1000  0.528895
2000-01-01   30  1000  0.949571
2000-01-02   30  1000  0.662527
2000-01-07   30  1000  0.013572
2000-01-08   30  1000  0.622846
2000-01-09   30  1000  0.673660
2000-01-10   30  1000  0.971945

В этом примере таблицы минимальная дата равна 2000-01-01, а максимальная - 2000-01-10, затем следует добавить следующие примеры:

2000-01-01   10  1000  NaN
2000-01-04   10  1000  NaN
2000-01-06   10  1000  NaN
2000-01-08   10  1000  NaN
2000-01-03   30  1000  NaN
2000-01-04   30  1000  NaN
2000-01-05   30  1000  NaN
2000-01-06   30  1000  NaN

То, как я делаю это прямо сейчас, - это создать новый фрейм данных, содержащий все минимальные дни, сделать максимум для каждой пары id C -idP и объединяя все в большой массив данных. Я соединяю исходный фрейм данных с большим фреймом данных on=['date', 'idC', 'idP'] how='outer'.

Этот метод быстрый, но занимает много памяти (для больших наборов данных), поскольку я в основном создаю большую копию оригинала dataframe. Можно ли получить этот результат с меньшим использованием памяти?

---

Мое текущее решение с использованием фиктивных данных:

def newcols(dframe, idC, idP):
    aux = dframe.copy()
    aux['idC'] = idC
    aux['idP'] = idP
    return aux


def get_date_df_full(start, final, idCs, idPs):
    all_dates = pd.DataFrame(pd.date_range(start, final, freq='D'), columns=['date'])
    all_dates['date'] = all_dates['date'].dt.strftime('%Y-%m-%d')

    all_dfs = pd.concat([newcols(all_dates, idC, idP) for idC in idCs for idP in idPs])

    date_df = spark.createDataFrame(all_dfs)
    date_df = date_df.withColumn('date', date_df.date.cast('date'))
    return date_df


data = spark.createDataFrame(df)

min_date = data.agg({'date': 'min'}).collect()[0][0]
max_date = data.agg({'date': 'max'}).collect()[0][0]

dates_full = get_date_df_full(min_date, max_date, idC_list, idP_list)
end_data = data.join(dates_full, on=['date', 'idC', 'idP'], how='outer')

---

Бонус Вопрос: Возможно ли добиться чего-то похожего на объясненное, но используя уникальный диапазон дат для каждой пары id C -idP? Или, может быть, уникальный диапазон дат для каждого idP?