Flux de travail "données importantes" à l'aide de pandas

je pense que les réponses ci-dessus manquent une approche simple que j'ai trouvé très utile.

quand j'ai un fichier trop gros pour être chargé en mémoire, je le décompose en plusieurs fichiers plus petits (par ligne ou cols)

exemple: en cas de valeur de 30 jours de données de trading de taille ~30Go, je le brise dans un fichier par jour de taille ~1Go. Par la suite, je traite chaque dossier séparément et je additionne les résultats à la fin de la période 151910920."

L'un des plus grands avantages est qu'il permet le traitement parallèle des fichiers (plusieurs threads ou processus)

l'autre avantage est que la manipulation de fichiers (comme Ajouter/Supprimer des dates dans l'exemple) peut être effectuée par des commandes shell régulières, ce qui n'est pas possible dans des formats de fichiers plus avancés/compliqués

cette approche ne couvre pas tous les scénarios, mais est très utile dans un grand nombre d'entre eux

si vos ensembles de données sont entre 1 et 20 Go, vous devriez obtenir une station de travail avec 48 Go de RAM. Pandas peut alors contenir L'ensemble des données en mémoire vive. Je sais que ce n'est pas la réponse que vous cherchez ici, mais faire de l'informatique scientifique sur un ordinateur portable avec 4 Go de mémoire vive n'est pas raisonnable.

il y a maintenant, deux ans après la question, un équivalent pandas "out-of-core": dask . Il est excellent! Bien qu'il ne supporte pas toutes les fonctionnalités de pandas, vous pouvez aller vraiment loin avec elle.

je sais que c'est un vieux fil, mais je pense que la bibliothèque Blaze mérite d'être vérifiée. C'est fait pour ce genre de situations.

à Partir de la documentation:

Blaze étend L'utilisabilité de NumPy et Pandas au calcul distribué et Hors-noyau. Blaze fournit une interface similaire à celle du DataFrame numpy ND-Array ou Pandas, mais établit une correspondance entre ces interfaces familières et une variété d'autres interfaces. des moteurs de calcul comme Postgres ou Spark.

Edit: Par ailleurs, il est pris en charge par ContinuumIO et Travis Oliphant, auteur de NumPy.

C'est le cas de pymongo. J'ai aussi prototypé en utilisant sql server, sqlite, HDF, ORM (SQLAlchemy) en python. D'abord et avant tout pymongo est un DB basé sur un document, donc chaque personne serait un document ( dict d'attributs). Beaucoup de gens forment une collection et vous pouvez avoir beaucoup de collections (personnes, bourse, revenu).

pd.dateframe - > pymongo Note: j'utilise le chunksize dans read_csv pour le conserver dans des enregistrements de 5 à 10k(pymongo laisse tomber la socket si en plus grand)

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

l'interrogation: gt = supérieur...

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

.find() retourne un itérateur donc j'utilise couramment ichunked pour couper en itérateurs plus petits.

Que Diriez-vous d'une jointure puisque j'obtiens Normalement 10 sources de données à coller ensemble:

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

puis (dans mon cas je dois parfois agg sur aJoinDF d'abord avant son "fusible".)

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

et vous pouvez alors écrire les nouvelles informations à votre collection principale via la méthode de mise à jour ci-dessous. (collecte logique vs sources de données physiques).

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

sur les plus petits lookups, juste dénormaliser. Par exemple, vous avez du code dans le document et vous n'avez qu'à ajouter le texte du code de zone et faire une recherche dict pendant que vous créez des documents.

Maintenant vous avez un bel ensemble de données basées sur une personne, vous pouvez libérer votre logique sur chaque cas et faire plus d'attributs. Enfin, vous pouvez lire dans pandas votre 3 à la mémoire max indicateurs clés et faire pivots / agg / exploration de données. Cela fonctionne pour moi, pour 3 millions d'enregistrements avec des nombres/texte/catégories/codes/flotteurs/...

vous pouvez également utiliser les deux méthodes intégrées dans MongoDB (MapReduce et aggregate framework). voir ici pour plus d'informations sur le cadre agrégé , car il semble être plus facile que MapReduce et semble pratique pour le travail agrégé rapide. Notez que je n'avais pas besoin de définir mes domaines ou relations, et je peux ajouter des éléments à un document. À l'état actuel de l'évolution rapide de numpy, pandas, Python toolset, MongoDB m'aide juste à me mettre au travail:)

j'ai repéré cela un peu tard, mais je travaille avec un problème similaire (modèles de remboursement anticipé d'hypothèque). Ma solution a été de sauter la couche pandas HDFStore et d'utiliser des pytables droits. Je sauve chaque colonne comme un tableau HDF5 individuel dans mon fichier final.

mon workflow de base est d'abord d'obtenir un fichier CSV à partir de la base de données. Je l'ai ouvert, donc il n'est pas aussi énorme. Puis je le convertis en un fichier HDF5 orienté ligne, en itérant dessus en python, en convertissant chaque ligne en données réelles taper, et l'écrire dans un fichier HDF5. Cela prend quelques dizaines de minutes, mais il n'utilise pas de mémoire, car il ne fonctionne que ligne par ligne. Puis j'ai "transposé" le fichier HDF5 orienté ligne dans un fichier HDF5 orienté colonne.

le tableau transpose ressemble à:

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

la relire ressemble alors à:

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)

maintenant, je cours généralement sur une machine avec une tonne de mémoire, donc je ne pourrais pas être assez prudent avec mon l'utilisation de la mémoire. Par exemple, par défaut, l'opération de chargement lit l'ensemble du jeu de données.

cela fonctionne généralement pour moi, mais c'est un peu lourd, et je ne peux pas utiliser la magie pytables Fantaisie.

Edit: le véritable avantage de cette approche, par rapport au tableau-of-records pytables par défaut, est que je peux ensuite charger les données dans R en utilisant h5r, qui ne peut pas gérer les tables. Ou, du moins, je n'ai pas pu l'amener à charger des tables hétérogènes.

une autre variation

beaucoup des opérations effectuées dans les pandas peuvent aussi être faites comme une requête db (sql, mongo)

L'utilisation D'un RDBMS ou mongodb vous permet d'effectuer certaines agrégations dans la requête DB (qui est optimisée pour les grandes données, et utilise le cache et les index efficacement)

plus tard, vous pouvez effectuer le post-traitement en utilisant pandas.

l'avantage de cette méthode est que vous gagnez la PD optimisations pour travailler avec de grandes données, tout en définissant la logique dans une syntaxe déclarative de haut niveau - et ne pas avoir à traiter les détails de décider ce qu'il faut faire en mémoire et ce qu'il faut faire hors du noyau.

et bien que le langage de requête et les pandas soient différents, il n'est généralement pas compliqué de traduire une partie de la logique de l'un à l'autre.

Un truc que j'ai trouvé utile pour les "grandes données" cas d'utilisation est de réduire le volume des données en réduisant la précision de 32 bits. Il n'est pas applicable dans tous les cas, mais dans de nombreuses applications, la précision 64 bits est excessive et les économies de mémoire 2x en valent la peine. Pour rendre un point évident encore plus évident:

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

tel que noté par d'autres, après quelques années un équivalent pandas" hors-de-cœur "a émergé: dask . Bien que DASK ne soit pas un remplacement drop-in de pandas et de toutes ses fonctionnalités, il se démarque pour plusieurs raisons:

Dask est une bibliothèque de calcul parallèle flexible pour le calcul analytique qui est optimisé pour la planification dynamique des tâches pour les charges de travail computationnelles interactives de Les collections de "Big Data" comme les tableaux parallèles, les images de données et les listes qui étendent les interfaces communes comme les itérateurs NumPy, Pandas ou Python à des environnements plus grands que la mémoire ou distribués et les échelles des ordinateurs portables aux clusters.

DASK souligne les vertus suivantes:

• Familier: Fournit parallélisée tableau NumPy et les Pandas DataFrame objets
• Flexible: fournit une interface de planification des tâches pour des charges de travail plus personnalisées et l'intégration avec d'autres projet.
• natif: permet le calcul distribué en Python pur avec accès à la pile PyData.
• Fast: fonctionne avec une faible charge, une faible latence, et une sérialisation minimale nécessaire pour les Algorithmes numériques rapides
• Echelles vers le haut: fonctionne avec résilience sur les grappes avec des milliers de noyaux Echelles vers le bas: Trivial pour mettre en place et exécuter sur un ordinateur portable dans un seul processus
• Responsive: Conçu avec l'informatique interactive à l'esprit, il fournit une rétroaction rapide et des diagnostics pour aider les humains

et pour ajouter un exemple de code simple:

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

remplace certains pandas comme celui-ci:

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

et, surtout digne de mention, fournit à travers le concurrent.les contrats à terme de l'interface générale pour la soumission de tâches personnalisées:

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

prendre en considération Ruffus si vous allez à la simple voie de la création d'un pipeline de données, qui est divisé en plusieurs petits fichiers.

il est intéressant de mentionner ici rayon ainsi,

c'est un cadre de calcul distribué, qui a sa propre implémentation pour pandas d'une manière distribuée.

il suffit de remplacer l'importation pandas, et le code devrait fonctionner comme il est:

# import pandas as pd
import ray.dataframe as pd

#use pd as usual

peut lire plus de détails ici:

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

j'ai récemment rencontré un problème similaire. J'ai simplement trouvé que lire les données en morceaux et les ajouter comme je l'écris en morceaux à la même csv fonctionne bien. Mon problème était d'ajouter une colonne date basée sur des informations dans un autre tableau, en utilisant la valeur de certaines colonnes comme suit. Cela peut aider ceux confondus par dask et hdf5, mais plus familier avec les pandas comme moi.

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)