Comment charger des gigaoctets de données de Google Cloud Storage dans une base de données pandas?

Question

J'essaie de charger des gigaoctets de données de Google Cloud Storage ou de Google BigQuery dans des données pandas afin que je puisse tenter d'exécuter OneClassSVM et Isolation Forest de scikit (ou toute autre classification unaire ou PU). J'ai donc essayé pandas-gbq mais de tenter d'exécuter

pd.read_gbq(query, 'my-super-project', dialect='standard') 

cause ma machine à elle sigkill quand il est seulement 30% complet. Et je ne peux pas le charger localement, et ma machine n'a pas assez d'espace et ne semble pas assez efficace.

J'ai aussi essayé

from google.cloud import storage 
client = storage.Client() 
bucket = client.get_bucket('bucket-id-here') 
# Then do other things... 
blob = bucket.get_blob('remote/path/to/file.txt') 
print(blob.download_as_string()) 

sur je peux charger 1/10 ou 1/5 de mes données disponibles, mais ma machine me dit finalement qu'il a manqué de mémoire.

TLDR: Est-il possible que je peux exécuter mon code personnalisé (avec numpy, pandas géants, et même tensorflow) dans le nuage ou un supercalculateur Farway où je peux facilement et charger efficacement les données de Google Cloud Storage ou Google BigQuery?

Answer 1

Je ne pense pas que vous alliez dans la bonne direction. Je vais essayer d'expliquer comment je travaille habituellement avec les données et j'espère que cela vous donne des idées.

J'ai d'abord tendance à travailler avec de petits ensembles de données en appliquant une technique d'échantillonnage ou en interrogeant pendant moins de jours. Dans cette étape, il est bon d'utiliser pandas ou d'autres outils développés pour les petites données et de construire des modèles, soulever des statistiques, trouver moments et ainsi de suite. Après avoir pris connaissance des données, je commence à travailler avec les outils Big Data.

Plus précisément, j'ai un très petit cluster Dataproc où j'ai déjà configuré un jupyter notebook pour exécuter le code pyspark.

La mémoire totale de votre cluster devra dépasser la mémoire totale que vous utilisez en entrée. L'utilisation de pandas ou de données d'étincelles devrait être simple pour vous, comme vous pouvez le voir dans cette section blog post par databricks, spark propose déjà cette fonctionnalité.

Ensuite, vient l'implémentation des algorithmes. Spark offre déjà built-in algorithms prêt-à-l'emploi, vous pouvez jouer avec eux. Si les algorithmes que vous souhaitez implémenter ne sont pas disponibles, vous pouvez émettre une requête dans leur référentiel ou la créer vous-même (vous pouvez utiliser l'implémentation Scipy de Python comme guide et la transposer dans l'environnement spark).

Voici un exemple de la façon dont je charge les données pour l'un des algorithmes que j'utilise pour construire un système de recommender pour notre entreprise:

from pyspark.sql import functions as sfunc 
from pyspark.sql import types as stypes 

schema = stypes.StructType().add("fv", stypes.StringType()).add("sku", stypes.StringType()).add("score", stypes.FloatType()) 
train_df = spark.read.csv('gs://bucket_name/pyspark/train_data*.gz', header=True, schema=schema) 

Spark distribuera automatiquement ces données à travers les différents travailleurs dont vous disposez dans votre grappe. Après cela, j'exécute principalement des requêtes et mappe/réduit les étapes pour obtenir des corrélations entre les skus.

En ce qui concerne la maintenance de votre code actuel, il ne s'agira probablement pas d'une mise à l'échelle pour les gros volumes de données.Vous pouvez néanmoins trouver beaucoup de ressources pour combiner la puissance de numpy avec spark, comme dans this example par exemple.