Как обработать датафрейм с миллиардами записей за считанные секунды?

Анализ больших данных в Python переживает эпоху возрождения. Она началась с библиотеки NumPy. Эта библиотека, кстати, является одной из составных частей тех инструментов, о которых пойдёт речь в этом материале. В 2006 году тема обработки больших данных постепенно набирала обороты, этот процесс ускорился с появлением Hadoop. Потом появилась библиотека pandas со своими структурами данных DataFrame, которые обычно называют просто «датафреймами». В 2014 году большие данные стали мейнстримом, в этом же году появилась платформа Apache Spark. В 2018 году вышла библиотека Dask и другие средства для анализа данных в Python.

^[1]

Каждый месяц мне попадаются новые инструменты для анализа данных в Python, которые мне очень хочется освоить. Потратив час-другой на их изучение, можно, в долгосрочной перспективе, сэкономить немало времени. Кроме того, важно следить за тем новым, что происходит в интересующей тебя сфере технологий. Возможно, вы полагаете, что эта статья будет посвящена библиотеке Dask ^[2]. Но это не так. Сегодня я расскажу вам об одной недавно обнаруженной мной Python-библиотеке, о которой стоит знать тем, кто занимается анализом данных.

Python-библиотека Vaex

Vaex ^[3] — это высокопроизводительная Python-библиотека, предназначенная для организации «ленивой» обработки датафреймов (похожих на датафреймы pandas), рассчитанная на визуализацию и исследование больших наборов табличных данных. Эта библиотека умеет вычислять основные статистические показатели по исследуемым данным. При этом обработка миллиарда записей может составлять что-то около секунды. Библиотека поддерживает множество средств визуализации, что помогает исследовать большие данные в интерактивном режиме.

Сравнение Vaex и Dask

Библиотека Vaex не похожа на Dask, но используемые в ней структуры данных похожи на сущности DataFrame, применяемые в Dask. Они построены на базе датафреймов pandas. Это означает, что Dask получает в наследство ^[4] от pandas определённые ограничения по работе с датафреймами. Например — требование, в соответствии с которыми данные, которые планируется обрабатывать, должны быть полностью загружены в память. В случае с Vaex это не так.

Vaex не делает копий датафреймов, в результате средствами этой библиотеки можно обрабатывать масштабные датафреймы на компьютерах с небольшим объёмом оперативной памяти.

И Vaex и Dask используют «ленивые» механизмы обработки данных. Основное различие между ними заключаться в том, что Vaex вычисляет значения полей тогда, когда они нужны, а в Dask надо явным образом использовать функцию вычисления значений.

Для того чтобы раскрыть весь потенциал Vaex нужно, чтобы данные были бы представлены в формате HDF5 или Apache Arrow.

Установка Vaex

Установить Vaex так же легко, как и любой другой Python-пакет:

pip install vaex

Эксперименты с Vaex

Создадим датафрейм pandas, в котором содержится 1 миллион строк и 1000 столбцов. Это позволит нам получить большой файл с данными.

import vaex
import pandas as pd
import numpy as np
n_rows = 1000000
n_cols = 1000
df = pd.DataFrame(np.random.randint(0, 100, size=(n_rows, n_cols)), columns=['col%d' % i for i in range(n_cols)])
df.head()

Вот как выглядят эти данные.

Данные для проведения эксперимента

Сколько оперативной памяти использует этот датафрейм?

df.info(memory_usage='deep')

Сведения об использовании памяти

Сохраним этот датафрейм на диск для того чтобы потом прочитать его с помощью Vaex.

file_path = 'big_file.csv'
df.to_csv(file_path, index=False)

Если напрямую прочитать этот CSV-файл с помощью Vaex, особых выгод мы от этого не получим. Скорость чтения будет похожей на скорость, обеспечиваемую pandas. На моём ноутбуке и Vaex и pandas читают эти данные примерно за 85 секунд.

Для того чтобы увидеть возможности Vaex, CSV-данные нужно преобразовать в формат HDF5 (Hierarchical Data Format version 5). В Vaex есть функция для выполнения такого преобразования, которая позволяет работать с данными, объём которых превышает объём доступной оперативной памяти. Это достигается благодаря тому, что исходные данные разбиваются на фрагменты.

Если не удаётся, из-за нехватки памяти, открыть достаточно большой файл с помощью pandas, этот файл можно преобразовать в формат HDF5 и обработать с помощью Vaex.

dv = vaex.from_csv(file_path, convert=True, chunk_size=5_000_000)

Вышеприведённая функция автоматически создаёт HDF5-файл и сохраняет его на диск.

Проверим тип переменной dv:

type(dv)
# вывод
vaex.hdf5.dataset.Hdf5MemoryMapped

Теперь прочитаем набор данных объёмом 7.5 Гб с помощью Vaex. Прямо сейчас нам его читать не придётся — дело в том, что его уже представляет переменная dv. Но следующую команду мы всё же запустим, сделав это для того чтобы узнать о том, какова скорость выполнения этой операции:

dv = vaex.open('big_file.csv.hdf5')

На выполнение этой команды у Vaex уходит меньше секунды. Но Vaex, на самом деле, из-за использования механизмов «ленивой» обработки данных, этот файл не читает.

Теперь давайте заставим Vaex прочитать файл, посчитав сумму значений в столбце col1:

suma = dv.col1.sum()
suma
# array(49486599)

А вот эта команда меня по-настоящему удивила. Для вычисления суммы Vaex тоже понадобилось меньше секунды. Как это возможно? А возможно это благодаря мэппингу памяти.

Визуализация данных

Vaex показывает хорошую скорость и на визуализации данных. В библиотеке есть особые функции: plot1d, plot2d и plot2d_contour.

dv.plot1d(dv.col2, figsize=(14, 7))

Результат визуализации данных

Виртуальные столбцы

Vaex, при добавлении к набору данных нового столбца, создаёт виртуальный столбец. Он не занимает память. Значения, хранящиеся в нём, вычисляются, что называется, «на лету».

dv['col1_plus_col2'] = dv.col1 + dv.col2
dv['col1_plus_col2']

Новый столбец

Эффективная фильтрация данных

Vaex не создаёт копий датафреймов при фильтрации данных. Это способствует более эффективному использованию памяти.

dvv = dv[dv.col1 > 90]

Агрегирование данных

В Vaex агрегирование данных работает немного не так, как в pandas. Но самое важное тут то, что в Vaex эта операция выполняется очень быстро.

Создадим новый столбец, содержащий результаты сравнения dv.col1 >= 50:

dv['col1_50'] = dv.col1 >= 50

Vaex комбинирует операции группировки и агрегирования данных в одной команде. Следующая команда группирует данные по столбцу col1_50 и вычисляет сумму столбца col3:

dv_group = dv.groupby(dv['col1_50'], agg=vaex.agg.sum(dv['col3']))
dv_group

Результаты выполнения команды

Соединения данных

Vaex позволяет соединять данные без создания копий данных в памяти. Это, как и другие возможности библиотеки, способствует экономному использованию памяти. Функция join, показанная ниже, покажется знакомой пользователям pandas:

dv_join = dv.join(dv_group, on=’col1_50')

Итоги

В этом материале мы рассмотрели основные особенности и возможности библиотеки Vaex. Если она вас заинтересовала — загляните в её репозиторий ^[5].

А вам пригодится библиотека Vaex?

Автор: ru_vds

Источник ^[6]