Обработка крупных наборов данных в Python: от Pandas до распределённых инструментов

Ниже приведены расширенные примеры, демонстрирующие нестандартные приёмы работы с большими данными.

Пример 1: Параллельная обработка чанков с помощью concurrent.futures

Вместо последовательного цикла можно распределить обработку чанков по нескольким процессам.

import pandas as pd
from concurrent.futures import ProcessPoolExecutor

def process_chunk(chunk):
    # Моделируем тяжёлую агрегацию
    return chunk[chunk['value'] > 50]['amount'].sum()

chunks = pd.read_csv('large.csv', chunksize=100000)

with ProcessPoolExecutor(max_workers=4) as executor:
    results = list(executor.map(process_chunk, chunks))

total = sum(results)
print(f"Итоговая сумма: {total}")

Итоговая сумма: 1234567890

Пояснение: каждый чанк обрабатывается отдельным процессом, что утилизирует несколько ядер. Важно убедиться, что функция process_chunk не требует много памяти на один вызов.

Пример 2: Создание виртуальных столбцов в Vaex для экономии памяти

Vaex позволяет добавлять вычисляемые столбцы без их материализации.

import vaex

df = vaex.open('sensor_data.hdf5')
df['temperature_f'] = df.temperature_c * 9/5 + 32
df['status'] = df.temperature_c > 100

print(df[['timestamp', 'temperature_c', 'temperature_f', 'status']].head(5))

  timestamp  temperature_c  temperature_f  status
0         1         98.0        208.4   False
1         2        102.5        216.5    True
2         3         99.9        211.8   False
3         4        105.0        221.0    True
4         5         97.8        208.0   False

Пояснение: столбцы temperature_f и status не занимают дополнительного места, так как являются виртуальными и вычисляются на лету.

Пример 3: Использование Dask для скользящего окна по большому датасету

Dask поддерживает оконные функции, например rolling mean, с автоматической сегментацией.

import dask.dataframe as dd

df = dd.read_csv('timeseries.csv', blocksize='50MB')
df = df.set_index('timestamp')
rolling_mean = df['value'].rolling(window=10).mean().compute()
print(rolling_mean.head(15))

timestamp
0         NaN
1         NaN
2         NaN
3         NaN
4         NaN
5         NaN
6         NaN
7         NaN
8         NaN
9         5.2
10        5.5
11        5.8
12        6.0
13        6.3
14        6.7
Name: value, dtype: float64

Пояснение: Dask разбивает DataFrame по времени (благодаря индексу) и обрабатывает каждую часть с перекрытием границ, чтобы окно работало корректно.

Пример 4: Чтение данных из нескольких файлов с помощью glob и Modin

Modin умеет автоматически распределять чтение множества CSV-файлов.

import modin.pandas as pd
import glob

files = glob.glob('data/2023/*.csv')
df = pd.concat([pd.read_csv(f) for f in files])

print(f"Загружено строк: {len(df)}")
print(df.groupby('region')['revenue'].sum())

Загружено строк: 45200000
region
North    5.123e8
South    3.789e8
East     2.456e8
West     4.567e8
Name: revenue, dtype: float64

Пояснение: Modin параллельно читает каждый файл, а объединённый DataFrame остаётся ленивым до момента вычисления.

Пример 5: Оптимизация памяти с помощью downcast в Pandas

Автоматическое понижение типов после загрузки снижает потребление.

import pandas as pd

df = pd.read_csv('sales.csv')
# Выбор только числовых столбцов
num_cols = df.select_dtypes(include=['int64', 'float64']).columns

for col in num_cols:
    df[col] = pd.to_numeric(df[col], downcast='integer' if df[col].dtype == 'int64' else 'float')

print(f"Память до: {df.memory_usage(deep=True).sum() / 1024**2:.1f} MB (после downcast)")

Память до: 124.3 MB (после downcast)

Пояснение: downcast автоматически выбирает минимальный подтип (int8, int16, float32 и т.д.), что особенно эффективно для столбцов с ограниченным диапазоном значений.