Что можно использовать как альтернативу PySpark?

Альтернативы PySpark для обработки больших данных

Введение

PySpark долгое время был индустриальным стандартом для распределённой обработки данных. Но за 10+ лет я видел, как появились более современные и часто более удобные альтернативы. Расскажу о практических вариантах.

Почему искать альтернативы PySpark?

# PySpark имеет проблемы:
# 1. Сложность: много кода для простых задач
# 2. Медленность: Питон + JVM = overhead
# 3. Отладка: трудно дебажить распределённый код
# 4. Стоимость: требует больших кластеров для эффективности

1. Dask (Python, самый близкий аналог)

Описание

Dask — это распределённые вычисления на Python, которые работают как на одной машине, так и на кластере.

import dask.dataframe as dd
import pandas as pd

# Читаем большой CSV
df_dask = dd.read_csv('huge_file.csv')

# Трансформация (lazy evaluation)
result = df_dask[
    (df_dask['age'] > 25) & (df_dask['salary'] > 50000)
][['name', 'salary']].groupby('department')['salary'].mean()

# Вычисление
result_computed = result.compute()
print(result_computed)

Плюсы

• Знакомый API (как pandas)
• Быстрее PySpark на средних объёмах
• Хорошо работает с NumPy, Pandas, Scikit-learn
• Легче дебажить
• Меньше оверхеда

Минусы

• Не масштабируется на петабайты (как Spark)
• Меньше ecosystem

Когда использовать

• Данные 10GB - 1TB
• Нужен быстрый разработка
• Работаешь с pandas-like API

2. Polars (Rust + Python, супер быстро)

Описание

Polars — это современная DataFrame библиотека, написанная на Rust. Намного быстрее pandas/Dask на некоторых операциях.

import polars as pl

# Читаем CSV
df = pl.read_csv('data.csv')

# Query (очень быстро)
result = (
    df
    .filter((pl.col('age') > 25) & (pl.col('salary') > 50000))
    .select(['name', 'salary'])
    .groupby('department')
    .agg(pl.col('salary').mean())
)

print(result)

# Для больших данных, использующих streaming
df_large = pl.scan_csv('huge_file.csv')
result = (
    df_large
    .filter(pl.col('salary') > 100000)
    .groupby('department')
    .agg(pl.col('salary').sum())
    .collect()  # Вычисление
)

Плюсы

• Очень быстрый (написан на Rust)
• Lazy evaluation и query optimization
• Удобный API
• Меньше памяти
• Параллелизм встроен

Минусы

• Новый (менее стабильный API)
• Меньше экосистема
• Нет native распределения (только для one machine)

Бенчмарки

import polars as pl
import pandas as pd
import dask.dataframe as dd
import time

data = {'a': range(10_000_000), 'b': range(10_000_000)}

# Pandas
start = time.time()
df_pd = pd.DataFrame(data)
result_pd = df_pd[df_pd['a'] > 5_000_000].groupby('b')['a'].sum()
print(f"Pandas: {time.time() - start:.2f}s")

# Dask
start = time.time()
df_dask = dd.from_pandas(pd.DataFrame(data), npartitions=4)
result_dask = df_dask[df_dask['a'] > 5_000_000].groupby('b')['a'].sum().compute()
print(f"Dask: {time.time() - start:.2f}s")

# Polars
start = time.time()
df_pl = pl.DataFrame(data)
result_pl = df_pl.filter(pl.col('a') > 5_000_000).groupby('b').agg(pl.col('a').sum())
print(f"Polars: {time.time() - start:.2f}s")

# Output (на моей машине):
# Pandas: 3.45s
# Dask: 1.23s
# Polars: 0.18s  <- 19x быстрее Pandas!

3. Apache Arrow + DuckDB (SQL, быстро)

Описание

DuckDB — это встроённая SQL база данных, оптимизированная для OLAP (аналитика). Работает с Parquet файлами.

import duckdb

# Запрос SQL к большому файлу (не загружает в памяти всё сразу)
result = duckdb.query("""
    SELECT department, AVG(salary) as avg_salary
    FROM 'data.parquet'
    WHERE age > 25 AND salary > 50000
    GROUP BY department
    ORDER BY avg_salary DESC
""").to_df()

print(result)

# С Pandas
df = duckdb.from_df(pd.DataFrame(data))
result = df.query("""
    SELECT * FROM df
    WHERE salary > 100000
""").to_df()

# С Parquet (очень быстро благодаря Parquet statistics)
result = duckdb.query("""
    SELECT department, COUNT(*), AVG(salary)
    FROM 'huge_data.parquet'
    WHERE date > '2023-01-01'
    GROUP BY department
""").to_df()

Плюсы

• Работает с Parquet файлами (очень быстро)
• SQL синтаксис (знаком большинству)
• Встроённый (no overhead)
• Оптимизирован для аналитики
• Можно читать из S3

Минусы

• Один узел (no native распределение)
• Не для production analytics (как Spark)

Когда использовать

• Данные в Parquet
• SQL запросы
• Ad-hoc аналитика
• Нужна скорость на one machine

4. Ray (Distributed Python, масштабируемо)

Описание

Ray — это система для распределённых вычислений на Python. Более гибкая, чем Spark.

import ray
import pandas as pd

ray.init()

# Распределённый Dataframe (RayDF)
from ray.data import read_parquet

ds = read_parquet('data.parquet')

# Трансформация (распределённо)
result = (
    ds
    .filter(lambda x: x['age'] > 25)
    .map_batches(lambda batch: batch[batch['salary'] > 50000])
    .groupby('department')
    .aggregate(
        avg_salary=('salary', 'mean'),
        count=('id', 'count')
    )
    .to_pandas()
)

print(result)

ray.shutdown()

Плюсы

• Гибче, чем Spark (не только DataFrames)
• Хорошо для ML pipelines
• Легче масштабировать
• Питон-нативный (не JVM overhead)

Минусы

• Менее зрелый, чем Spark
• Меньше документации

Когда использовать

• ML pipelines (hyperparameter tuning, training)
• Batch processing
• Нужна гибкость

5. Pandas + Multiprocessing / ThreadPoolExecutor

Описание

Для данных до 100GB можно просто использовать Pandas с параллелизацией в Python.

import pandas as pd
from concurrent.futures import ProcessPoolExecutor
import os

def process_chunk(file_path):
    """Обработка одного файла"""
    df = pd.read_csv(file_path)
    df = df[df['salary'] > 50000]
    return df.groupby('department')['salary'].mean()

# Список файлов
files = [f'data/part_{i}.csv' for i in range(100)]

# Параллельная обработка
with ProcessPoolExecutor(max_workers=8) as executor:
    results = executor.map(process_chunk, files)

# Объединяем результаты
final_result = pd.concat(results, axis=1).mean(axis=1)
print(final_result)

Плюсы

• Простота
• Нет оверхеда (чистый Python)
• Легко дебажить
• Достаточно для многих задач

Минусы

• Не масштабируется на петабайты
• Нет оптимизации на уровне системы

6. Vaex (Out-of-memory DataFrames)

Описание

Vaex работает с DataFrames, которые не помещаются в памяти (memory-mapped files).

import vaex

# Читает большой CSV как memory-mapped (не загружает всё в ОЗУ)
df = vaex.open('huge_data.csv')

# Операции выполняются лениво (lazy)
df_filtered = df[(df.age > 25) & (df.salary > 50000)]
df_grouped = df_filtered.groupby('department', agg={'salary': 'mean'})

# Вычисление
result = df_grouped.to_pandas()
print(result)

Плюсы

• Работает с файлами больше, чем ОЗУ
• Быстрая фильтрация
• Удобный API

Минусы

• Меньше операций поддерживается
• Менее популярен

7. Ibis (Universal DataFrame Language)

Описание

Ibis — это абстракция над разными backends. Один код работает на Pandas, Dask, Polars, DuckDB и даже SQL базах.

import ibis

# Выбираем backend
con = ibis.duckdb.connect('data.db')

# Пишем запрос (будет трансляться в SQL)
table = con.table('employees')
result = (
    table
    .filter([table.age > 25, table.salary > 50000])
    .group_by('department')
    .aggregate(avg_salary=table.salary.mean())
)

# Выполняем
df = result.execute()
print(df)

# Тот же код работает на разных backend'ах!
con_polars = ibis.polars.connect({'employees': pl.DataFrame(data)})
df_polars = result.to_pandas()  # будет использовать Polars

Плюсы

• Один код на разные backend'ы
• Хорошая абстракция
• Новый стандарт (как LINQ для SQL)

Минусы

• Ещё развивается
• Может быть медленнее, чем native API

Сравнительная таблица

Инструмент | Масштаб | Скорость | Простота | Когда использовать
─────────────────────────────────────────────────────────────────────
Pandas | <100GB | Медленно | Очень | Отладка, малые данные
Dask | 100GB-1TB | Средне | Хорошо | Средние данные, Pandas API
Polars | <1TB | Очень | Хорошо | Максимальная скорость
DuckDB | <1TB | Очень | Хорошо | SQL, Parquet файлы
Ray | 1TB+ | Хорошо | Средне | ML pipelines, гибкость
PySpark | 1TB+ | Хорошо | Сложно | Петабайты, кластеры
Vaex | >ОЗУ | Средне | Средне | Out-of-memory данные
Ibis | varies | Хорошо | Хорошо | Multi-backend код

Практический совет: Выбор стека

Если данные < 10GB

# Просто используй Pandas
import pandas as pd

df = pd.read_csv('data.csv')
result = df.groupby('department')['salary'].mean()

Если данные 10GB - 1TB

# Используй Polars или DuckDB
import polars as pl

df = pl.read_csv('data.csv')
result = df.groupby('department').agg(pl.col('salary').mean())

Если данные > 1TB и много машин

# Только тогда используй Spark (или Ray)
from pyspark.sql import SparkSession

spark = SparkSession.builder.getOrCreate()
df = spark.read.csv('data.csv')
result = df.groupBy('department').agg({'salary': 'mean'})

Заключение

PySpark когда-то был единственным выбором. Сейчас есть много альтернатив:

• Polars — если нужна максимальная скорость
• DuckDB — если работаешь с SQL и Parquet
• Dask — если нужен Pandas API
• Ray — для ML pipelines
• PySpark — только если действительно петабайты данных

Мой совет: начни с Polars/DuckDB. Переходи на Spark только если действительно упрёшься в лимиты.