"Duże przepływy danych" za pomocą pand

Myślę, że w powyższych odpowiedziach brakuje prostego podejścia, które uznałem za bardzo przydatne.

Gdy mam plik, który jest zbyt duży, aby załadować go do pamięci, dzielę plik na wiele mniejszych plików (albo przez wiersz lub cols)

Przykład: w przypadku 30-dniowego handlu danymi o rozmiarze ~30GB, rozkładam je na plik dziennie o rozmiarze ~1GB. Następnie przetwarzam każdy plik osobno i agreguję wyniki na końcu

Jedną z największych zalet jest to, że umożliwia równoległe przetwarzanie plików (wiele wątków lub procesów)

Inną zaletą jest to, że manipulowanie plikami (jak dodawanie/usuwanie dat w przykładzie) może być wykonywane przez zwykłe polecenia powłoki, co nie jest możliwe w bardziej zaawansowanych/skomplikowanych formatach plików

To podejście nie obejmuje wszystkich scenariuszy, ale jest bardzo przydatne w wielu z nich

Jest teraz, dwa lata po pytaniu, "poza rdzeniem" pandy odpowiednik: dask . To jest doskonałe! Chociaż nie obsługuje wszystkich funkcjonalności pand, możesz zajść naprawdę daleko. Aktualizacja: w ciągu ostatnich dwóch lat został on konsekwentnie utrzymywany i istnieje znaczna społeczność użytkowników pracujących z Dask.

A teraz, cztery lata po pytaniu, pojawia się kolejny wysokowydajny odpowiednik "Out-of-core" w Vaex . "Wykorzystuje mapowanie pamięci, zero Polityka kopiowania pamięci i leniwe obliczenia dla najlepszej wydajności (brak marnowania pamięci)."Może obsługiwać zbiory danych o miliardach wierszy i nie zapisuje ich w pamięci (co umożliwia nawet analizę na nieoptymalnym sprzęcie).

Jeśli Twoje zestawy danych mają od 1 do 20 GB, powinieneś otrzymać stację roboczą z 48 GB PAMIĘCI RAM. Wtedy Pandy mogą trzymać cały zestaw danych w pamięci RAM. Wiem, że nie jest to odpowiedź, której szukasz tutaj, ale robienie obliczeń naukowych na notebooku z 4GB PAMIĘCI RAM nie jest rozsądne.

Wiem, że to stary wątek, ale myślę, że biblioteka Blaze jest warta sprawdzenia. Jest stworzony do tego typu sytuacji.

Z docs:

Blaze rozszerza użyteczność NumPy i pandy na komputery rozproszone i poza rdzeniem. Blaze zapewnia interfejs podobny do interfejsu NumPy ND-Array lub Panda DataFrame, ale mapuje te znane interfejsy na wiele innych silników obliczeniowych, takich jak Postgres lub Spark.

Edit: Nawiasem mówiąc, jest wspierany przez ContinuumIO i Travisa Oliphanta, autora NumPy.

Tak jest w przypadku pymongo. Mam również prototypów przy użyciu SQL server, SQLite, HDF, ORM (SQLAlchemy) w Pythonie. Pymongo jest przede wszystkim dokumentem bazującym na DB, więc każda osoba byłaby dokumentem (dict atrybutów). Wiele osób tworzy kolekcję i możesz mieć wiele kolekcji (ludzie, Giełda, dochód).

Pd.dateframe - > pymongo Uwaga: używam chunksize in read_csv, aby utrzymać go do 5 do 10K rekordów (pymongo zrzuca Gniazdo, jeśli jest większe)

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

Querying: gt = większe niż...

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

.find() zwraca iterator, więc często używam ichunked do siekania na mniejsze Iteratory.

A co powiesz na join skoro normalnie dostaję 10 źródeł danych do wklejenia razem:

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

Następnie (w moim przypadku czasami muszę się agg na aJoinDF najpierw przed jego "mergeable".)

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

A następnie możesz zapisać nowe informacje do swojej głównej kolekcji za pomocą metody aktualizacji poniżej. (zbiór logiczny a fizyczne źródła danych).

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

Na mniejszych Szukaj, po prostu denormalizuj. Na przykład, masz kod w dokumencie i po prostu dodaj tekst kodu pola i wykonaj wyszukiwanie dict podczas tworzenia dokumentów.

Teraz masz ładny zbiór danych oparty na osobie, możesz uwolnić swoją logikę na każdym przypadku i zrobić więcej atrybutów. Wreszcie można odczytać w pandy swoje 3 do pamięci max kluczowych wskaźników i zrobić pivots / agg / eksploracji danych. Działa mi to na 3 miliony rekordów z numerami / dużym tekstem/kategoriami/kodami/pływakami/...

Ty może również używać dwóch metod wbudowanych w MongoDB (MapReduce i aggregate framework). zobacz tutaj, aby uzyskać więcej informacji o frameworku agregatowym , ponieważ wydaje się być łatwiejszy niż MapReduce i wygląda przydatnie do szybkiej pracy agregatowej. Zauważ, że nie musiałem definiować moich pól ani relacji i mogę dodawać elementy do dokumentu. W obecnym stanie szybko zmieniającego się zestawu narzędzi numpy, pandas, python, MongoDB pomaga mi po prostu dostać się do pracy:) {]}

Jeden trik, który znalazłem pomocne dla dużych danych przypadków użycia jest zmniejszenie objętości danych poprzez zmniejszenie precyzji float do 32-bitowych. Nie ma to zastosowania we wszystkich przypadkach, ale w wielu aplikacjach 64-bitowa precyzja to przesada i warto mieć 2x oszczędności pamięci. Aby jeszcze bardziej oczywista sprawa:

>>> 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

Zauważyłem to trochę późno, ale pracuję z podobnym problemem(modele przedpłat hipotecznych). Moim rozwiązaniem było pominięcie warstwy Pandy HDFStore i użycie prostych pytabli. Zapisuję każdą kolumnę jako pojedynczą tablicę HDF5 w moim ostatecznym pliku.

Mój podstawowy obieg pracy polega na pobraniu pliku CSV z bazy danych. I gzip to, więc nie jest tak ogromny. Następnie konwertuję to do pliku HDF5 zorientowanego na wiersz, iterując go w Pythonie, konwertując każdy wiersz na prawdziwy typ danych i zapisując go do Plik HDF5. Zajmuje to kilkadziesiąt minut, ale nie używa żadnej pamięci, ponieważ działa tylko wiersz po wierszu. Następnie "transponuję" plik HDF5 zorientowany wierszem do pliku HDF5 zorientowanego kolumną.

Tabela wygląda następująco:

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()

Odczytanie tego z powrotem wygląda następująco:

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)

To ogólnie działa dla mnie, ale jest trochę niezgrabne, i nie mogę użyć fantazyjnej magii pytables.

Edit: prawdziwą zaletą tego podejścia, nad domyślną tablicą rekordów, jest to, że mogę następnie załadować dane do R za pomocą h5r, który nie radzi sobie z tabelami. A przynajmniej nie byłem w stanie załadować heterogenicznych tabel.

Jak zauważyli inni, po kilku latach pojawił się odpowiednik "Out-of-core" pand: dask . Mimo że dask nie jest zamiennikiem pand i cała jego funkcjonalność wyróżnia się z kilku powodów:]}

Dask jest elastyczną biblioteką obliczeń równoległych do obliczeń analitycznych, która jest zoptymalizowana pod kątem dynamicznego harmonogramowania zadań dla interaktywnych obciążeń obliczeniowych Zbiory "Big Data", takie jak tablice równoległe, ramki danych i listy, które rozszerzają wspólne interfejsy, takie jak Iteratory NumPy, Pandas lub Python do środowisk większych niż pamięć lub rozproszonych i skalują się od laptopów do klastrów.

Dask podkreśla następujące cnoty:

• Familiar: dostarcza paralelized numpy array i pandy DataFrame obiektów
• Elastyczny: zapewnia interfejs harmonogramowania zadań dla większej liczby niestandardowych obciążeń i integracji z innymi projektami.
• natywny: umożliwia przetwarzanie rozproszone w czystym Pythonie z dostępem do Pydata stack.
Fast: działa z niskim obciążeniem, małym opóźnieniem i minimalną serializacją niezbędną dla szybkich algorytmów numerycznych]}
• skalowalne: działa stabilnie na klastrach z 1000 rdzeniami skalowane w dół: trywialne jest skonfigurowanie i uruchomienie na laptopie w jednym procesie
Responsywny: zaprojektowany z myślą o interaktywnych komputerach, zapewnia szybkie informacje zwrotne i diagnostykę, aby pomóc ludziom]}

I dodać prosty kod próbka:

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

Zastępuje jakiś kod pandy taki:

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

I, szczególnie godne uwagi, zapewnia poprzez interfejs concurrent.futures ogólną infrastrukturę do składania zadań niestandardowych:

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()

Warto tutaj wspomnieć Ray jak również
jest to rozproszony framework obliczeniowy, który posiada własną implementację dla pand w sposób rozproszony.

Wystarczy wymienić import pandas, a kod powinien działać tak jak jest:

# import pandas as pd
import ray.dataframe as pd

#use pd as usual

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

Jeszcze jedna odmiana

Wiele operacji wykonywanych w Pandzie może być również wykonanych jako zapytanie db (sql, mongo)

Później możesz wykonać post processing używając pand.

Zaletą tej metody jest to, że uzyskujesz optymalizacje DB do pracy z dużymi danymi, jednocześnie definiując logikę w składnia deklaratywna na wysokim poziomie - bez konieczności zajmowania się szczegółami decydowania, co robić w pamięci, a co robić poza rdzeniem.

I chociaż język zapytań i pandy są różne, zwykle nie jest skomplikowane tłumaczenie części logiki z jednego na drugi.

Rozważ Ruffus jeśli wybierzesz prostą ścieżkę tworzenia potoku danych, który jest podzielony na wiele mniejszych plików.

Chciałbym zwrócić uwagę na pakiet Vaex.

Vaex jest biblioteką Pythona dla leniwych ramek danych (podobnych do pand), do wizualizacji i eksploracji dużych zbiorów danych tabelarycznych. Może obliczać statystyki, takie jak średnia, suma, liczba, odchylenie standardowe itp., na siatce n-wymiarowej do miliarda (10⁹) obiektów / wierszy na sekundę. Wizualizacja odbywa się za pomocą histogramów, Wykresów gęstości i renderowania objętości 3d, umożliwiając interaktywną eksplorację dużych danych. Vaex wykorzystuje pamięć mapowanie, polityka zerowej kopii pamięci i leniwe obliczenia dla najlepszej wydajności (brak marnowania pamięci).

Zajrzyj do dokumentacji: https://vaex.readthedocs.io/en/latest / API jest bardzo zbliżone do API pand.

Ostatnio natknąłem się na podobny problem. Znalazłem po prostu czytanie danych w kawałkach i dołączanie ich, gdy piszę je w kawałkach do tego samego pliku csv działa dobrze. Moim problemem było dodanie kolumny daty na podstawie informacji w innej tabeli, używając wartości niektórych kolumn w następujący sposób. Może to pomóc tym, którzy są zdezorientowani przez dask i hdf5, ale bardziej zaznajomieni z pandami takimi jak ja.

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)

W tej chwili pracuję " jak " ty, tylko na mniejszą skalę, dlatego nie mam PoC dla mojej sugestii.

Wydaje mi się jednak, że odnajduję sukces w użyciu pickle ' a jako systemu buforowania i outsourcingu wykonywania różnych funkcji do plików-wykonywanie tych plików z mojego commando / main file; na przykład używam prepare_use.py aby przekonwertować typy obiektów, podziel zestaw danych na zestaw danych testowych, walidacyjnych i predykcyjnych.

Jak działa buforowanie z pickle? Używam sznurków w celu uzyskania dostępu do pickle-plików, które są dynamicznie tworzone, w zależności od tego, jakie parametry i zbiory danych zostały przekazane (z tym staram się przechwycić i określić, czy program był już uruchomiony, za pomocą .shape dla zestawu danych, dict dla przekazywanych parametrów). Szanując te środki, dostaję ciąg, aby spróbować znaleźć i przeczytać .pickle-plik i może, jeśli zostanie znaleziony, pominąć czas przetwarzania, aby przejść do wykonania, nad którym obecnie pracuję.

Korzystając z baz danych napotkałem podobne problemy, czyli dlaczego jednak cieszy mnie korzystanie z tego rozwiązania - na pewno jest wiele ograniczeń - na przykład przechowywanie ogromnych zestawów ogórków ze względu na redundancję. Aktualizacja tabeli z przed do po transformacji może być wykonywana z odpowiednim indeksowaniem - Walidacja informacji otwiera całą inną książkę (próbowałem skonsolidować dane indeksowane i przestałem korzystać z bazy danych po 2 godzinach zasadniczo - jak chciałbym skakać z powrotem po każdym procesie transformacji)

Mam nadzieję, że moje 2 grosze pomogą Ci w jakimś sposób.

Pozdrawiam.

Dlaczego Pandy ? Czy próbowałeś standardowego Pythona?

Użycie biblioteki standardowej Pythona. Pandas jest przedmiotem częstych aktualizacji, nawet w ostatnim wydaniu stabilnej wersji.

Używając standardowej biblioteki Pythona Twój kod będzie zawsze uruchamiany.

Jednym ze sposobów na zrobienie tego jest wyobrażenie sobie, w jaki sposób chcesz, aby Twoje dane były przechowywane i jakie pytania chcesz rozwiązać dotyczące danych. Następnie narysuj schemat organizowania danych (tabel myślowych), które pomoże Ci wyszukać dane, niekoniecznie normalizację.

Możesz dobrze wykorzystać:

Lista słowników do przechowywania danych w pamięci (Think Amazon EC2) lub na dysku, jeden dict to jeden wiersz,
• generatory do przetwarzania danych wiersz po wierszu, aby nie przepełniać pamięci RAM,
• zrozumienie listy do zapytania danych,
• użyj Counter, DefaultDict, ...
• przechowuj dane na dysku twardym, używając dowolnego wybranego rozwiązania do przechowywania, json może być jednym z nich.

Ram i HDD stają się z czasem coraz tańsze, a standardowy python 3 jest powszechnie dostępny i stabilny.

Fondamental pytanie, które próbujesz rozwiązać, to " jak odpytywać duże zbiory danych ?". Architektura hdfs jest mniej więcej tym, co tutaj opisuję (modelowanie danych z danymi zapisywanymi na dysku).

Załóżmy, że masz 1000 petabajtów danych, nie ma mowy, że będziesz mógł je przechowywać w Dasku lub pandach, twoje największe szanse oto, aby przechowywać go na dysku i przetwarzać go za pomocą generatorów.