Масштабирование больших наборов данных

Введение

В этом практическом занятии (лабораторной работе) рассматривается вопрос о том, как масштабировать анализ данных для более крупных наборов данных с использованием библиотеки pandas. Здесь рассматриваются методы, такие как загрузка меньшего объема данных, использование эффективных типов данных, разбиение на части (chunking) и использование других библиотек, таких как Dask. Важно отметить, что pandas лучше всего подходит для аналитики, выполняемой в памяти, и может не быть наилучшим инструментом для очень больших наборов данных.

Советы по виртуальной машине (VM)

После завершения запуска виртуальной машины нажмите в левом верхнем углу, чтобы переключиться на вкладку Notebook и получить доступ к Jupyter Notebook для практики.

Иногда вам может потребоваться подождать несколько секунд, пока Jupyter Notebook полностью загрузится. Валидация операций не может быть автоматизирована из-за ограничений Jupyter Notebook.

Если вы столкнетесь с проблемами во время обучения, не стесняйтесь обращаться к Labby. После занятия оставьте отзыв, и мы оперативно решим проблему для вас.

Генерация набора данных

Первый шаг - сгенерировать большой набор данных для тестирования. Мы создаем набор данных с большим количеством столбцов, который можно сохранить в файле формата parquet. Для этого шага требуются библиотеки pandas и numpy.

import pandas as pd
import numpy as np

def make_timeseries(start="2000-01-01", end="2000-12-31", freq="1D", seed=None):
    ## Функция для генерации временных рядов
    index = pd.date_range(start=start, end=end, freq=freq, name="timestamp")
    n = len(index)
    state = np.random.RandomState(seed)
    columns = {
        "name": state.choice(["Alice", "Bob", "Charlie"], size=n),
        "id": state.poisson(1000, size=n),
        "x": state.rand(n) * 2 - 1,
        "y": state.rand(n) * 2 - 1,
    }
    df = pd.DataFrame(columns, index=index, columns=sorted(columns))
    if df.index[-1] == end:
        df = df.iloc[:-1]
    return df

timeseries = [
    make_timeseries(freq="1T", seed=i).rename(columns=lambda x: f"{x}_{i}")
    for i in range(10)
]
ts_wide = pd.concat(timeseries, axis=1)
ts_wide.to_parquet("timeseries_wide.parquet")

Загрузка меньшего объема данных

Вместо загрузки всех данных мы можем загрузить только нужные нам столбцы. Здесь мы демонстрируем два метода для загрузки меньшего объема данных из файла формата parquet.

## Вариант 1: Загрузить все данные, а затем отфильтровать
columns = ["id_0", "name_0", "x_0", "y_0"]
pd.read_parquet("timeseries_wide.parquet")[columns]

## Вариант 2: Загрузить только запрошенные столбцы
pd.read_parquet("timeseries_wide.parquet", columns=columns)

Использование эффективных типов данных

По умолчанию pandas использует типы данных, которые не всегда являются наиболее экономными по памяти. В этом шаге показано, как использовать более эффективные типы данных для хранения больших наборов данных в памяти.

ts = make_timeseries(freq="30S", seed=0)
ts.to_parquet("timeseries.parquet")
ts = pd.read_parquet("timeseries.parquet")

## Преобразование столбца 'name' в тип 'category' для повышения эффективности
ts2 = ts.copy()
ts2["name"] = ts2["name"].astype("category")

## Сокращение числовых столбцов до наименьших типов
ts2["id"] = pd.to_numeric(ts2["id"], downcast="unsigned")
ts2[["x", "y"]] = ts2[["x", "y"]].apply(pd.to_numeric, downcast="float")

Использование разбиения на части (chunking)

Разбиение на части (chunking) - это метод, который позволяет разбить большую задачу на меньшие задачи, которые можно решать независимо. Пока каждая часть помещается в память, вы можете работать с наборами данных, которые намного больше объема памяти.

files = pathlib.Path("data/timeseries/").glob("ts*.parquet")
counts = pd.Series(dtype=int)
for path in files:
    df = pd.read_parquet(path)
    counts = counts.add(df["name"].value_counts(), fill_value=0)
counts.astype(int)

Использование других библиотек

Другие библиотеки, такие как Dask, могут обрабатывать наборы данных, размер которых превышает объем памяти. Dask предоставляет API, похожее на API pandas, и может обрабатывать данные параллельно.

import dask.dataframe as dd

ddf = dd.read_parquet("data/timeseries/ts*.parquet", engine="pyarrow")

## Вычисление частот значений с использованием Dask
ddf["name"].value_counts().compute()

Резюме

В этом практическом занятии (lab) мы продемонстрировали различные методы масштабирования анализа данных для работы с большими наборами данных с использованием pandas. Мы сгенерировали большой набор данных, научились загружать меньшее количество данных, использовать эффективные типы данных и применять разбиение на части (chunking). Мы также рассмотрели, как использовать другие библиотеки, такие как Dask, для обработки наборов данных, размер которых превышает объем памяти. Методы и концепции, изученные в этом практическом занятии, будут полезны при работе с большими наборами данных в проектах по анализу данных.