Chargement des DataFrames Pandas dans QuestDB

Dans cet article, découvrez comment améliorer votre capacité d’analyse de séries chronologiques en utilisant le package QuestDB Python pour ingérer des Pandas DataFrames.

Introduction

Pandas est une bibliothèque open source d’analyse et de manipulation de données pour Python qui est devenue un outil essentiel pour les data scientists et les analystes. Il fournit un moyen simple et intuitif de manipuler les données, ce qui en fait un choix populaire pour les tâches d’analyse de données.

Cependant, bien que Pandas soit excellent pour les ensembles de données de petite à moyenne taille, il peut avoir du mal avec de grands ensembles de données qui dépassent la mémoire disponible de la machine sur laquelle il s’exécute. C’est là que QuestDB excelle, conçu spécifiquement pour les opérations hautes performances dans de tels scénarios, ce qui en fait la solution incontournable pour les tâches d’analyse de données exigeantes.

En chargeant Pandas DataFrames dans QuestDB, nous pouvons tirer parti du puissant capacités de traitement des données de la base de données, vous permettant d’adapter vos opérations d’analyse et de manipulation de données à de grands ensembles de données. Nous apprendrons comment charger de grands Pandas Dataframes dans QuestDB. Nous utilisons les enregistrements de trajets en taxi jaunes et verts publiés par la NYC Taxi and Limousine Commission comme source de données.

Dans ce didacticiel, nous allons apprendre à charger de grandes dataframes Pandas dans QuestDB. Nous utilisons les enregistrements de trajets en taxi jaunes et verts publiés par la NYC Taxi and Limousine Commission comme source de données.

Conditions préalables

Pour ce tutoriel, il est recommandé d’avoir une compréhension de base de Python et SQL. De plus, vous devrez avoir les éléments suivants installés sur votre machine :

Obtenir les données à ingérer

Avant de commencer à charger les données dans QuestDB, nous devons obtenir les données avec lesquelles nous allons travailler. Comme mentionné ci-dessus, nous utiliserons les enregistrements de trajets en taxi jaune et vert du NYC TLC. Téléchargeons les données :

• Créez un nouveau répertoire appelé pandas-to-questdb et un data répertoire à l’intérieur de celui-ci.
• Editez et exécutez la commande suivante dans votre terminal pour télécharger les fichiers parquet :

curl -L -o ./data/yellow_tripdata_2022-<MONTH>.parquet https://d37ci6vzurychx.cloudfront.net/trip-data/yellow_tripdata_2022-<MONTH>.parquet

Assurez-vous de remplacer <MONTH> avec le numéro préfixé par zéro du mois que vous souhaitez télécharger (entre 01 et 11, le 12ème mois n’est pas disponible au moment de la rédaction).

Maintenant, nous avons les données à ingérer. Il est temps d’essayer de le charger avec Pandas.

Chargement d’enregistrements en mémoire

Vous avez peut-être déjà remarqué que les fichiers téléchargés sont au format Parquet. Parquet est un format de stockage en colonne couramment utilisé pour le traitement du Big Data. Ils sont optimisés pour une utilisation avec les moteurs de traitement de données volumineux modernes et offrent un stockage et une récupération efficaces des données par rapport aux formats de stockage traditionnels basés sur des lignes tels que CSV et JSON.

Avant de pouvoir charger des données, nous allons mettre en place un environnement de production de simulation où nous pourrons facilement tester ce qui se passe si Pandas ne peut pas charger les fichiers Parquet en mémoire. En production, nous rencontrons souvent des situations où nous devons faire face à des contraintes de mémoire, et cet environnement peut refléter cela.

Exécutez la commande suivante pour créer un nouveau conteneur Docker exécuté avec une limite de mémoire de 1 Gio. Si le conteneur atteint cette limite, Docker le tuera, ou le système d’exploitation OOM tuera le processus que nous exécutons :

docker run -it -m 1g -v "$(pwd)":/tutorial -w /tutorial --net host python:3.11.1-slim-bullseye /bin/bash

Maintenant, nous avons un conteneur Docker Python 3.11 basé sur Ubuntu. Installons nos exigences. Créer un requirements.txt fichier avec le contenu ci-dessous :

pandas>=1.5.3
psycopg[binary]>=3.1.8
pyarrow>=11.0.0
questdb>=1.1.0

Maintenant, exécutez pip install -r requirements.txt à l’intérieur du conteneur. Pip installera les exigences Python.

À ce stade, nous avons un environnement de test dans lequel nous pouvons charger les données. Créez un nouveau fichier appelé data_loader.py avec le contenu suivant :

# data_loader.py
import pandas as pd

df = pd.read_parquet("./data/yellow_tripdata_2022-01.parquet")
print(df.head())

Maintenant, exécutez-le dans le conteneur Docker en exécutant python data_loader.py. Le programme s’exécute avec succès et nous devrions voir ce qui suit :

Nous venons de charger les enregistrements de trajets en taxi pour janvier 2022 ! Essayons de charger plus de données. Remplacer le contenu de data_loader.py avec le code ci-dessous pour charger tous les fichiers du répertoire de données et exécuter à nouveau le programme :

# data_loader.py

import os
import glob
import pandas as pd

records = glob.glob(os.path.join("data", "*.parquet"))

df = pd.concat((pd.read_parquet(r) for r in records), ignore_index=True)

print(df.head())

Lors de l’exécution de la data_loader.pyvous devriez recevoir un message d’erreur : « Tué.” Comme vous pouvez le supposer, le tueur OOM a mis fin au processus. Nous n’avons pas pu charger l’ensemble de données ; par conséquent, nous ne pouvons pas travailler avec cela. Nous avons besoin d’une approche différente.

Ingérer dans QuestDB

Dans une nouvelle fenêtre de terminal, démarrez un conteneur QuestDB en exécutant :

docker run --rm -it -p 8812:8812 -p 9009:9009 --net host --name questdb questdb/questdb

La base de données est maintenant prête à recevoir les données. Mettre à jour le data_loader.py pour ingérer des données dans QuestDB en utilisant le questdb package qui utilise le protocole de ligne InfluxDB (ILP) sur TCP pour un débit maximal.

Pour gérer de grands ensembles de données, nous allons lire les fichiers un par un et transférer leur contenu vers QuestDB. Ensuite, nous utiliserons QuestDB pour interroger les données et charger les résultats dans Pandas DataFrames. Refactorisez le chargeur de données en fonction de ce qui précède :

# data_loader.py

import os
import glob
import pandas as pd
from questdb.ingress import Sender

def main():
   files = glob.glob(os.path.join("data", "*.parquet"))

   with Sender("127.0.0.1", 9009) as sender:
       for file in files:
           df = pd.read_parquet(file)
           print(f"ingesting {len(df.index)} rows from {file}")
           sender.dataframe(df, table_name="trips", at="tpep_pickup_datetime")

if __name__ == "__main__":
   main()

Commençons depuis le début. Le premier changement majeur que vous remarquerez est que nous devons spécifier le nom d’hôte et le numéro de port dans le script pour l’exécuter.

Ensuite, nous parcourons les fichiers Parquet et les chargeons en mémoire à l’aide de Pandas. Après cela, en utilisant le client Python de QuestDB, nous ingérons QuestDB directement à partir de Pandas DataFrames.

Dans le conteneur Python, exécutez python data_loader.py. Le script ingérera un fichier parquet à la fois.

Utilisation des données de trajet

Jusqu’à présent, nous avons préparé le jeu de données et l’avons chargé dans QuestDB. Il est temps d’exécuter quelques requêtes et de charger le résultat dans DataFrames. En utilisant l’ensemble des données, nous voulons savoir quel a été le montant total moyen payé par les passagers regroupés par les passagers.

Créez un nouveau fichier, appelé query_amount.py avec le contenu suivant :

# query_amount.py

import pandas as pd
import psycopg

QUERY = """
SELECT passenger_count, avg(total_amount)
   FROM 'trips'
   WHERE passenger_count > 0
   GROUP BY passenger_count
"""

if __name__ == "__main__":
   conn = psycopg.connect(
       dbname="questdb",
       host="127.0.0.1",
       user="admin",
       password="quest",
       port=8812,
   )

   df = pd.read_sql_query(QUERY, conn)

   print(df.head(10))

Semblable au script du chargeur de données, ce script nécessite l’hôte et le port. Dans le script ci-dessus, nous utilisons le client PostgreSQL Python et nous nous connectons à QuestDB à l’aide de celui-ci. Dans le conteneur Python, exécutez Python query_amount.py: