Кто-нибудь знает, как устранить ошибку «java .lang.OutOfMemoryError: G C overhead limit over Превышен» при работе с pyspark?

Question

Я написал приложение spark, которое запрашивает все данные из таблицы улья (100M строк, ~ 3 ГБ), собирает все эти данные обратно в драйвер через преобразование их во фрейм данных pandas, просматривает каждую запись и выполняет расчет и сохраняет результат в новом поле. Выполнение в отношении подмножества данных завершается нормально, но при выполнении в отношении совокупности возникает ошибка «java .lang.OutOfMemoryError: G C overhead limit over Превышен». Драйвер имеет более чем достаточно памяти для обработки этих данных, а также ограниченное количество копий. Драйвер имеет около 125 ГБ доступной памяти, и я установил в драйвере память на 32 ГБ. Кто-нибудь может помочь?

if __name__ == "__main__":
    spark = SparkSession.builder.appName('some_app').getOrCreate()

    sdf = spark.sql("SELECT * from some_database.some_table") # input
    df = sdf.toPandas() # input converted
    results = pd.DataFrame() # output

    for i in df.some_field.unique():
        df2 = "apply some logic"
        results = results.append(df2) # store processed data in results

    sdf2 = spark.createDataFrame(results)
    sdf2.registerTempTable("sdf2")
    spark.sql("DROP TABLE IF EXISTS some_database.my_results")
    spark.sql("CREATE TABLE IF NOT EXISTS some_database.my_results STORED AS PARQUET as select * from sdf2")

Answer 1

Как говорится в сообщении об ошибке, эта операция занимает слишком много памяти, поскольку данные 3 ГБ считываются в драйвер. Реальный объем памяти будет намного больше, чем 3 ГБ, я полагаю, 5-10X. Это не подходящий дизайн, поскольку он зависит от вычислительной мощности одного драйвера.

Вы должны создать кадр данных на основе таблицы Hive "some_database.some_table" без преобразования в pandas.