Рабочие потоки "больших данных" с использованием панд

Я думаю, что ответы выше отсутствуют простой подход, который я нашел очень полезным.

когда у меня есть файл, который слишком велик для загрузки в память, я разбиваю файл на несколько файлов меньшего размера (либо по строке, либо по cols)

пример: в случае 30-дневного объема торговых данных размером ~30 Гб я разбиваю его на файл в день размером ~1 ГБ. Впоследствии я обрабатываю каждый файл отдельно и агрегирую результаты в конце

одно из самых больших преимуществ что он позволяет параллельную обработку файлов (либо несколько потоков или процессов)

другое преимущество заключается в том, что манипуляции с файлами (например, добавление / удаление дат в примере) могут выполняться обычными командами оболочки, что невозможно в более продвинутых/сложных форматах файлов

этот подход не охватывает все сценарии, но очень полезен во многих из них

Если ваши наборы данных находятся между 1 и 20 ГБ, вы должны получить рабочую станцию с 48 ГБ оперативной памяти. Тогда панды могут держать весь набор данных в оперативной памяти. Я знаю, что это не тот ответ, который вы ищете здесь, но делать научные вычисления на ноутбуке с 4 ГБ ОЗУ не разумно.

теперь, через два года после вопроса, есть эквивалент "вне ядра" панды:ДАСК. Это превосходно! Хотя он не поддерживает все функции pandas, вы можете получить действительно далеко с ним.

Я знаю, что это старый нить, но я думаю Blaze библиотека стоит проверить. Он построен для таких ситуаций.

документы:

Blaze расширяет удобство использования NumPy и Pandas для распределенных и внеядерных вычислений. Blaze предоставляет интерфейс, аналогичный интерфейсу NumPy ND-Array или Pandas DataFrame, но отображает эти знакомые интерфейсы на множество других вычислительных механизмов, таких как Postgres или Искра.

Edit: кстати, его поддерживают ContinuumIO и Трэвис Олифант, автор NumPy.

это дело для pymongo. Я также прототипировал с помощью sql server, sqlite, HDF, ORM (SQLAlchemy) в python. В первую очередь pymongo-это БД на основе документов, поэтому каждый человек будет документом (dict атрибуты). Многие люди формируют коллекцию, и у вас может быть много коллекций (люди, фондовый рынок, доход).

pd.dateframe - > pymongo Примечание: я использую chunksize на read_csv чтобы сохранить его до 5 до 10k записей(pymongo сбрасывает сокет, если крупнее)

aCollection.insert((a[1].to_dict() for a in df.iterrows()))

запрос: gt = больше, чем...

pd.DataFrame(list(mongoCollection.find({'anAttribute':{'$gt':2887000, '$lt':2889000}})))

.find() возвращает итератор, поэтому я обычно использую ichunked нарезать на более мелкие итераторы.

как насчет соединения, так как я обычно получаю 10 источников данных для вставки вместе:

aJoinDF = pandas.DataFrame(list(mongoCollection.find({'anAttribute':{'$in':Att_Keys}})))

затем (в моем случае иногда приходится агг на aJoinDF сначала перед его "слиянием".)

df = pandas.merge(df, aJoinDF, on=aKey, how='left')

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

collection.update({primarykey:foo},{key:change})

при меньшем поиске просто денормализуйте. Например, у вас есть код в документе, и вы просто добавляете текст кода поля и делаете dict поиск при создании документов.

теперь у вас есть хороший набор данных, основанный на человеке, вы можете раскрыть свою логику в каждом случае и сделать больше атрибутов. Наконец, вы можете прочитать в панд ваши 3 в память Макс ключевых показателей и сделать повороты/agg/исследования данных. Это работает для меня для 3 миллионов записей с номерами / большим текстом/категориями/кодами/поплавками/...

вы также можете использовать два метода, встроенные в MongoDB (MapReduce и aggregate framework). смотрите здесь для получения дополнительной информации о aggregate framework, так как это кажется проще, чем MapReduce и выглядит удобно для быстрой агрегатной работы. Обратите внимание, что мне не нужно было определять мои поля или отношения, и я могу добавлять элементы в документ. В текущем состоянии быстро меняется numpy, панды, python toolset, MongoDB помогает мне просто приступить к работе:)

Я заметил это немного поздно, но я работаю с аналогичной проблемой (модели предоплаты ипотеки). Мое решение состояло в том, чтобы пропустить слой pandas HDFStore и использовать прямые pytables. Я сохраняю каждый столбец как отдельный массив HDF5 в моем окончательном файле.

мой основной рабочий процесс, чтобы сначала получить CSV-файл из базы данных. Я gzip его, так что это не так огромно. Затем я конвертирую это в ориентированный на строку файл HDF5, повторяя его в python, Преобразуя каждую строку в реальный тип данных и записывая это к файлу HDF5. Это занимает несколько десятков минут, но он не использует никакой памяти, так как он работает только строка за строкой. Потом "перенести" с построчным HDF5 файл в столбец-ориентированной HDF5 файл.

таблица транспонирования выглядит так:

def transpose_table(h_in, table_path, h_out, group_name="data", group_path="/"):
    # Get a reference to the input data.
    tb = h_in.getNode(table_path)
    # Create the output group to hold the columns.
    grp = h_out.createGroup(group_path, group_name, filters=tables.Filters(complevel=1))
    for col_name in tb.colnames:
        logger.debug("Processing %s", col_name)
        # Get the data.
        col_data = tb.col(col_name)
        # Create the output array.
        arr = h_out.createCArray(grp,
                                 col_name,
                                 tables.Atom.from_dtype(col_data.dtype),
                                 col_data.shape)
        # Store the data.
        arr[:] = col_data
    h_out.flush()

чтение его обратно в то выглядит так:

def read_hdf5(hdf5_path, group_path="/data", columns=None):
    """Read a transposed data set from a HDF5 file."""
    if isinstance(hdf5_path, tables.file.File):
        hf = hdf5_path
    else:
        hf = tables.openFile(hdf5_path)

    grp = hf.getNode(group_path)
    if columns is None:
        data = [(child.name, child[:]) for child in grp]
    else:
        data = [(child.name, child[:]) for child in grp if child.name in columns]

    # Convert any float32 columns to float64 for processing.
    for i in range(len(data)):
        name, vec = data[i]
        if vec.dtype == np.float32:
            data[i] = (name, vec.astype(np.float64))

    if not isinstance(hdf5_path, tables.file.File):
        hf.close()
    return pd.DataFrame.from_items(data)

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

это обычно работает для меня, но это немного неуклюже, и я не могу использовать причудливую магию pytables.

Edit: реальное преимущество этого подхода, по сравнению с массивом записей pytables по умолчанию, заключается в том, что я могу загрузить данные в R с помощью h5r, который не может обрабатывать таблицы. Или, по крайней мере, я не смог заставить его загружать разнородные таблицы.

еще одна вариация

многие операции, выполняемые в pandas, также могут быть выполнены как запрос БД (sql, mongo)

использование СУБД или mongodb позволяет выполнять некоторые агрегации в запросе БД (который оптимизирован для больших данных и эффективно использует кэш и индексы)

позже вы можете выполнить постобработку с помощью панд.

преимущество этого метода заключается в том, что вы получаете оптимизацию БД для работы с большими данными, в то время как все еще определяя логику в декларативном синтаксисе высокого уровня - и не имея дела с деталями решения, что делать в памяти и что делать вне ядра.

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

один трюк, который я нашел полезным для случаев использования" больших данных",-это уменьшить объем данных, уменьшив точность float до 32-бит. Это применимо не во всех случаях, но во многих приложениях 64-битная точность является излишней, и экономия памяти 2x стоит того. Чтобы сделать очевидный момент еще более очевидным:

>>> df = pd.DataFrame(np.random.randn(int(1e8), 5))
>>> df.info()
<class 'pandas.core.frame.DataFrame'>
RangeIndex: 100000000 entries, 0 to 99999999
Data columns (total 5 columns):
...
dtypes: float64(5)
memory usage: 3.7 GB

>>> df.astype(np.float32).info()
<class 'pandas.core.frame.DataFrame'>
RangeIndex: 100000000 entries, 0 to 99999999
Data columns (total 5 columns):
...
dtypes: float32(5)
memory usage: 1.9 GB

как отмечают другие, через несколько лет появился эквивалент "вне ядра" панд:ДАСК. Хотя dask не является выпадающей заменой панд и вся его функциональность выделяется по нескольким причинам:

Dask-это гибкая параллельная вычислительная библиотека для аналитических вычислений, оптимизированная для динамического планирования задач для интерактивных вычислительных нагрузок Коллекции "больших данных", такие как параллельные массивы, фреймы данных и списки, расширяющие общие интерфейсы, такие как итераторы NumPy, Pandas или Python для больших, чем память, или распределенных сред и масштабируются от ноутбуков до кластеров.

ДАСК подчеркивает следующие достоинства:

• знакомый: предоставляет распараллеленный массив NumPy и объекты DataFrame Pandas
• гибкий: обеспечивает интерфейс планирования задач для более пользовательских рабочих нагрузок и интеграции с другими проектами.
• родной: включает распределенный вычисления в чистом Python с доступом к стеку PyData.
• Fast: работает с низкими накладными расходами, низкой задержкой и минимальной сериализацией, необходимой для быстрых численных алгоритмов
• масштабирование: работает упруго на кластерах с 1000 ядер масштабирование вниз: тривиально настроить и запустить на ноутбуке в одном процессе
• отзывчивый: разработанный с учетом интерактивных вычислений он обеспечивает быструю обратную связь и диагностику, чтобы помочь люди

и добавить простой пример кода:

import dask.dataframe as dd
df = dd.read_csv('2015-*-*.csv')
df.groupby(df.user_id).value.mean().compute()

заменяет некоторые панды такой код:

import pandas as pd
df = pd.read_csv('2015-01-01.csv')
df.groupby(df.user_id).value.mean()

и, что особенно примечательно, обеспечивает через параллельный.фьючерсы интерфейс, общий для представления пользовательских задач:

from dask.distributed import Client
client = Client('scheduler:port')

futures = []
for fn in filenames:
    future = client.submit(load, fn)
    futures.append(future)

summary = client.submit(summarize, futures)
summary.result()

рассмотрим Ruffus Если вы идете по простому пути создания конвейера данных, который разбит на несколько небольших файлов.

здесь стоит упомянуть Рэй также,
это распределенная вычислительная структура, которая имеет свою собственную реализацию для панд распределенным способом.

просто замените импорт панд, и код должен работать как есть:

# import pandas as pd
import ray.dataframe as pd

#use pd as usual

можно прочитать более подробную информацию здесь:

https://rise.cs.berkeley.edu/blog/pandas-on-ray/

недавно я столкнулся с подобной проблемой. Я обнаружил, что просто читаю данные в кусках и добавляю их, когда я пишу их в кусках в тот же csv, хорошо работает. Моя проблема заключалась в добавлении столбца даты на основе информации в другой таблице, используя значение определенных столбцов следующим образом. Это может помочь тем, кто смущен dask и hdf5, но более знаком с пандами, такими как я.

def addDateColumn():
"""Adds time to the daily rainfall data. Reads the csv as chunks of 100k 
   rows at a time and outputs them, appending as needed, to a single csv. 
   Uses the column of the raster names to get the date.
"""
    df = pd.read_csv(pathlist[1]+"CHIRPS_tanz.csv", iterator=True, 
                     chunksize=100000) #read csv file as 100k chunks

    '''Do some stuff'''

    count = 1 #for indexing item in time list 
    for chunk in df: #for each 100k rows
        newtime = [] #empty list to append repeating times for different rows
        toiterate = chunk[chunk.columns[2]] #ID of raster nums to base time
        while count <= toiterate.max():
            for i in toiterate: 
                if i ==count:
                    newtime.append(newyears[count])
            count+=1
        print "Finished", str(chunknum), "chunks"
        chunk["time"] = newtime #create new column in dataframe based on time
        outname = "CHIRPS_tanz_time2.csv"
        #append each output to same csv, using no header
        chunk.to_csv(pathlist[2]+outname, mode='a', header=None, index=None)