Экономия RAM с pandas.read_sql

Рассмотрим потребление оперативной памяти (RAM) при загрузке pandas DataFrame из базы данных (БД).

Для этого воспользуемся мониторингом ресурсов на базе библиотеки psutil. Мониторинг фиксирует pss («Proportional Set Size») память, т. е. физически занимаемую процессом.

Используем небольшую таблицу синтетических транзакций:

Выгрузим DataFrame из БД, применив стандартный метод pandas.read_sql:

import time
import gc
import pandas as pd
import sqlalchemy as sa

connection_str = 'postgresql://db_user:db_pass@0.0.0.0:8032/postgres'
engine = sa.create_engine(connection_str)
query = f"""
SELECT
    check_id,
    store_id,
    sale_date,
    user_id,
    product_id,
    region,
    macro_region,
    quantity,
    category
FROM dummy_db."transactions_TEST";
"""
df = pd.read_sql(sql=query, con=engine)
gc.collect()
time.sleep(30)

# Сохраним DataFrame в файл:
df.to_csv("test_dataframe.csv", index=False)

Выгрузим DataFrame из сохраненного файла.

df = pd.read_csv("test_dataframe.csv")
gc.collect()
time.sleep(30)

Как видим загрузка одно и того же DataFrame из БД и из файла приводит к разнице в потреблении RAM в 5 раз. При выгрузке данных из файла — объем памяти в большей мере соизмерим с объемом данных в БД.

Почему же при выгрузке DataFrame из БД занимается столько лишней памяти?

На самом деле не при выгрузке из БД занимается лишняя память, а при загрузке из файла оптимизируется потребление памяти. При выгрузке DataFrame из файла происходит оптимизация интернированием.

Рассмотрим кол‑во уникальных значений в каждом столбце и кол‑во уникальных адресов в памяти для варианта с выгрузкой из БД:

print("Колонка         Кол-во уникальных значений     Кол-во уникальных адресов")
for column in df:
    print(column.ljust(15), str(df[column].nunique()).ljust(30), df[column].apply(id).nunique())

Рассмотрим кол-во уникальных значений в каждом столбце и кол-во уникальных адресов в памяти для варианта с выгрузкой из файла:

При выгрузке DataFrame из БД каждое значение занимает свой адрес в памяти даже несмотря на то, что это одно и то же значение.

При выгрузке DataFrame из файла занимается меньше уникальных адресов в памяти — происходит интернирование. Одинаковые значения не занимают несколько адресов в памяти.

Попробуем применить интернирование при выгрузке DataFrame из БД. Для этого будем выгружать данные чанками и каждый раз заменять одинаковые значения значениями, ссылающимися на одно и то же место в памяти. Но для этого нужно предварительно составить словарь с уникальными значениями, которые хранятся по уникальным адресам.

Выгрузка производится чанками, поскольку если выгрузить весь DataFrame и затем его оптимизировать, то пиковое потребление RAM останется на том же уровне, т. е. эффекта не будет.

connection_str = 'postgresql://db_user:db_pass@0.0.0.0:8032/postgres'
engine = sa.create_engine(connection_str)

# Создаем словарь с уникальными значениями из колонки:
intern_mapping_dict = {}
for column in ["check_id", "store_id", "user_id", "product_id", "region", "macro_region", "category"]:
    query = f"""
    SELECT {column} FROM dummy_db."transactions_TEST"
    GROUP BY {column} HAVING COUNT({column}) > 1
    """
    col_values = pd.read_sql(sql=query, con=engine)[column]
    print(column, len(col_values))
    intern_mapping_dict[column] = {v: v for v in col_values}

query = f"""
SELECT
    check_id,
    store_id,
    sale_date,
    user_id,
    product_id,
    region,
    macro_region,
    quantity,
    category
FROM dummy_db."transactions_TEST";
"""
df = pd.DataFrame()
for chunk in pd.read_sql(
    sql=query, con=engine, chunksize=50_000
):
    # Заполняем значениями, ссылающимися на одно и то же место в пямяти:
    for column, mapping in intern_mapping_dict.items():
        chunk[column] = chunk[column].map(lambda x: mapping.get(x, x))
    df = pd.concat([df, chunk], ignore_index=True)

gc.collect()
time.sleep(30)

Рассмотрим кол‑во уникальных значений в каждом столбце и кол‑во уникальных адресов в памяти для этого варианта выгрузки:

print("Колонка         Кол-во уникальных значений     Кол-во уникальных адресов")
for column in df:
    print(column.ljust(15), str(df[column].nunique()).ljust(30), df[column].apply(id).nunique())

Как видим теперь кол‑во уникальных значений для строковых типов соответствует кол‑ву уникальных адресов в памяти. То есть одни и те же значения теперь не занимают несколько ячеек памяти, а только по одной ячейке на одно уникальное значение.

В данном игрушечном примере мы сэкономили порядка 60% RAM.

Используя описанный метод оптимизации можно достичь экономии RAM до 90%!

В реальном проекте можно сократить потребление RAM со 100 GB до 10 GB.

Экономия зависит от кол‑ва повторяющихся строковых значений.

Обычно рекомендуют использовать категориальный тип для экономии ресурсов, но применение категориального типа обладает некоторыми недостатками:

1. Категориальный тип может «слететь» при некоторых операциях: concat, merge, join

2. При работе с категориальным типом могут возникать проблемы при использовании некоторых операций (например, группировки)

3. Категориальный тип создает дополнительную таблицу маппинга и хранит индексы, которые занимают дополнительную память. При применении интернирования не создается дополнительных таблиц маппинга.

Итог по потреблению RAM при различных способах выгрузки одного и того же DataFrame:

При способе с выгрузкой из БД с интернированием стремились получить значения по RAM такие же как при выгрузке из csv‑файла, но получили в 2 раза больше.

Проделаем еще один эксперимент — удалим данные сразу после выгрузки для всех трех случаев.

Обычная выгрузка из БД:

connection_str = 'postgresql://db_user:db_pass@0.0.0.0:8032/postgres'
engine = sa.create_engine(connection_str)
query = f"""
SELECT
    check_id,
    store_id,
    sale_date,
    user_id,
    product_id,
    region,
    macro_region,
    quantity,
    category
FROM dummy_db."transactions_TEST";
"""
df = pd.read_sql(sql=query, con=engine)
gc.collect()
time.sleep(30)
del df
gc.collect()
time.sleep(30)

Выгрузка из файла:

df = pd.read_csv("test_dataframe.csv")
gc.collect()
time.sleep(30)
del df
gc.collect()
time.sleep(30)

Выгрузка из БД с интернированием:

connection_str = 'postgresql://db_user:db_pass@0.0.0.0:8032/postgres'
engine = sa.create_engine(connection_str)

# Создаем словарь с уникальными значениями из колонки:
intern_mapping_dict = {}
for column in ["check_id", "store_id", "user_id", "product_id", "region", "macro_region", "category"]:
    query = f"""
    SELECT {column} FROM dummy_db."transactions_TEST"
    GROUP BY {column} HAVING COUNT({column}) > 1
    """
    col_values = pd.read_sql(sql=query, con=engine)[column]
    print(column, len(col_values))
    intern_mapping_dict[column] = {v: v for v in col_values}

query = f"""
SELECT
    check_id,
    store_id,
    sale_date,
    user_id,
    product_id,
    region,
    macro_region,
    quantity,
    category
FROM dummy_db."transactions_TEST";
"""
df = pd.DataFrame()
for chunk in pd.read_sql(
    sql=query, con=engine, chunksize=50_000
):
    # Заполняем значениями, ссылающимися на одно и то же место в пямяти:
    for column, mapping in intern_mapping_dict.items():
        chunk[column] = chunk[column].map(lambda x: mapping.get(x, x))
    df = pd.concat([df, chunk], ignore_index=True)

gc.collect()
time.sleep(30)
del df, col_values, chunk
del intern_mapping_dict, mapping
gc.collect()
time.sleep(30)

Итог по потреблению RAM при различных способах выгрузки одного и того же DataFrame:

Как видим в числом виде размер DataFrame при выгрузке из БД с интернированием имеет наименьший объем.

Остаток после удаления — это выгруженные из БД сырые данные, которые оставляет SQLAlchemy. Их к сожалению не очистить — это потеря RAM на время работы скрипта.

Итог

При выгрузке данных из БД с помощью pandas, применяя предложенный метод оптимизации памяти через интернирование, можно экономить до 90% RAM, что существенно при работе с большими данными.