Polars для обработки JSON и Parquet

Привет, Хабр!

Сегодня рассмотрим тему обработки временных рядов с помощью Polars.

Почему groupby_dynamic() лучше resample() из Pandas

Начну с того, что в Pandas для агрегации временных рядов принято использовать метод resample(). Он удобен и привычен, но имеет свои ограничения по производительности и гибкости. Polars, в свою очередь, имеет метод groupby_dynamic(), который позволяет группировать данные по динамическим временным интервалам.

Рассмотрим, как можно сгруппировать данные с часовыми метками по дневным интервалам:

import polars as pl from datetime import datetime, timedelta  # Генерируем данные: неделя записей с часовым интервалом dates = [datetime(2025, 1, 1) + timedelta(hours=i) for i in range(24 * 7)] values = [i % 5 + 1 for i in range(24 * 7)] df = pl.DataFrame({"timestamp": dates, "value": values})  # Группируем данные по дням и агрегируем: суммируем и считаем среднее resampled = df.groupby_dynamic("timestamp", every="1d").agg([     pl.col("value").sum().alias("daily_sum"),     pl.col("value").mean().alias("daily_mean") ]) print(resampled)

С помощью groupby_dynamic() определяем временной интервал (every=»1d») и сразу же агрегируем данные. Метод компилируется в код на Rust, что даёт прирост производительности по сравнению с Pandas. Если сравнить с Pandas, то код будет выглядеть примерно так:

import pandas as pd  df_pandas = pd.DataFrame({"timestamp": dates, "value": values}) df_pandas.set_index("timestamp", inplace=True) daily = df_pandas.resample("D").agg({"value": ["sum", "mean"]}) print(daily)

Rolling Windows: rolling_mean() и rolling_std() без overhead

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

Polars предлагает функции rolling_mean() и rolling_std(), которые оптимизированы до предела. Они реализованы на Rust

Пример вычисления скользящего среднего и стандартного отклонения:

df = pl.DataFrame({     "timestamp": dates,     "value": values }).with_columns([     pl.col("value").rolling_mean(window_size=24, min_periods=1).alias("rolling_mean"),     pl.col("value").rolling_std(window_size=24, min_periods=1).alias("rolling_std") ])  print(df.head(30))

Рассчитываем 24-часовое скользящее среднее и стандартное отклонение. Параметр min_periods=1 позволяет начать вычисления уже с первого значения, а оптимизированная реализация гарантирует, что даже при миллионах записей задержек практически не будет.

Также все это можно комбинировать с другими методами Polars.

Интерполяция пропущенных значений: Polars.interpolate()

Работая с временными рядами, мы часто сталкиваемся с пробелами в данных. От сбоя датчиков до ошибок при сборе информации — пропуски неизбежны. Хорошая новость: Polars предлагает метод interpolate() для быстрого заполнения пропущенных значений.

Создадим DataFrame с пропущенными значениями и применим линейную интерполяцию:

df_missing = pl.DataFrame({     "timestamp": dates,     "value": [None if i % 10 == 0 else i % 5 + 1 for i in range(len(dates))] })  # Применяем линейную интерполяцию для заполнения пропусков df_interpolated = df_missing.with_columns([     pl.col("value").interpolate(method="linear").alias("value_interpolated") ])  print(df_interpolated.head(30))

interpolate() позволяет указать метод интерполяции (в данном случае — «linear»).

Оптимизация анализа с помощью LazyFrame

Polars поддерживает ленивые вычисления через объект LazyFrame. Можно писать длинные цепочки преобразований, а Polars сам оптимизирует план выполнения и выполняет только необходимые вычисления в самый последний момент.

Пример оптимизированной обработки данных:

# Преобразуем DataFrame в LazyFrame lazy_df = df.lazy()  # Строим цепочку операций: фильтрация, группировка по дням, агрегирование result = lazy_df.filter(pl.col("value") > 2)\                 .groupby_dynamic("timestamp", every="1d")\                 .agg([                     pl.col("value").mean().alias("daily_mean")                 ])\                 .collect()  # Выполняем вычисления print(result)

Суть в том, что до вызова .collect() никаких вычислений не происходит. Это позволяет оптимизировать запрос, минимизировать избыточные проходы по данным и сократить время выполнения.

Пример обработкт временных рядов в магазине

Представим, магазина собирает данные о продажах с интервалом в час. Данные собираются постоянно, но иногда происходят сбои, и в ряде случаев значения пропадают. Задача — собрать данные, заполнить пропуски, агрегировать продажи по дням и вычислить скользящую недельную среднюю. Всё это — на Polars, и всё это должно работать быстро.

Сначала сгенерируем данные с часовыми записями за 30 дней, включая случайные пропуски:

import random random.seed(42)  # Создаем список дат: 30 дней, 24 записи в день dates_shop = [datetime(2021, 6, 1) + timedelta(hours=i) for i in range(30 * 24)] # Генерируем данные по продажам: случайные числа, пропуски ~10% sales = [random.randint(50, 200) if random.random() > 0.1 else None for _ in range(len(dates_shop))]  df_shop = pl.DataFrame({     "timestamp": dates_shop,     "sales": sales })

Заполним пробелы в данных, используя линейную интерполяцию:

df_shop = df_shop.with_columns([     pl.col("sales").interpolate(method="linear").alias("sales_interpolated") ])

Теперь агрегируем данные по дням, чтобы получить суммарные продажи и среднее значение за день:

daily_sales = df_shop.groupby_dynamic("timestamp", every="1d").agg([     pl.col("sales_interpolated").sum().alias("daily_total_sales"),     pl.col("sales_interpolated").mean().alias("daily_avg_sales") ])

Чтобы отследить тренды и сезонные колебания, посчитаем скользящую среднюю продаж за последние 7 дней:

daily_sales = daily_sales.with_columns([     pl.col("daily_total_sales").rolling_mean(window_size=7, min_periods=1).alias("weekly_sales_avg") ]) print(daily_sales)

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

В завершение напоминаю об открытых уроках, которые пройдут в Otus в марте:

• 25 марта: «Метрики и Prometheus».
  Узнать подробнее

• 27 марта: «PostgreSQL на стероидах: большие данные, высокие нагрузки и масштабирование без боли».
  Узнать подробнее

Больше актуальных навыков по аналитике данных вы можете получить в рамках практических онлайн-курсов от экспертов отрасли.