Filtrage de données Parquet avec Pandas

En ce qui concerne le filtrage de données à partir de fichiers Parquet à l’aide de pandas, plusieurs stratégies peuvent être mises en œuvre. Bien qu’il soit largement reconnu que la partition des données peut considérablement améliorer l’efficacité des opérations de filtrage, il existe d’autres méthodes pour optimiser la performance de la requête des données stockées dans des fichiers Parquet. La partition est simplement l’une des options.

Filtrage par Champs Partitionnés

Comme mentionné précédemment, cette approche n’est pas seulement la plus familière mais aussi généralement la plus impactante en termes d’optimisation des performances. La raison en est simple. Lorsque des partitions sont utilisées, il devient possible d’exclure la nécessité de lire des fichiers entiers ou même des répertoires entiers de fichiers (alias, déplacement de prédicat), ce qui entraîne une amélioration considérable et spectaculaire des performances.

import pandas as pd
import time
from faker import Faker
 
fake = Faker()
 
MIL=1000000
NUM_OF_RECORDS=10*MIL
FOLDER="/tmp/out/"
PARTITIONED_PATH=f"{FOLDER}partitioned_{NUM_OF_RECORDS}/"
NON_PARTITIONED_PATH_PREFIX=f"{FOLDER}non_partitioned_{NUM_OF_RECORDS}.parquet"
 
print(f"Creating fake data")
data = {
    'id': range(NUM_OF_RECORDS),  # Generate IDs from 1 to 100
    'name': [fake.name() for _ in range(NUM_OF_RECORDS)],
    'age': [fake.random_int(min=18, max=99) for _ in range(NUM_OF_RECORDS)],
    'state': [fake.state() for _ in range(NUM_OF_RECORDS)],
    'city': [fake.city() for _ in range(NUM_OF_RECORDS)],
    'street': [fake.street_address() for _ in range(NUM_OF_RECORDS)]
}
 
df = pd.DataFrame(data)
 
# écriture sans partitions
df.to_parquet(path=NON_PARTITIONED_PATH)
 
# écriture de données partitionnées
df.to_parquet(path=PARTITIONED_PATH, partition_cols=['state'])
 
# lecture de données non partitionnées
start_time = time.time()
df1 = pd.read_parquet(path=NON_PARTITIONED_PATH)
df1 = df1[df1['state']=='California']
runtime1 = (time.time()) - start_time  # 37 sec
 
# lecture de données partitionnées
start_time = time.time()
df2 = pd.read_parquet(path=PARTITIONED_PATH, filters=[('state','==','California')])
runtime2 = (time.time()) - start_time # 0.20 sec

L’amélioration du temps (ainsi que la réduction de la consommation de mémoire et de CPU) est considérable, passant de 37 secondes à seulement 0,20 seconde.

Filtrage par Champs Non-Partitionnés

Dans l’exemple ci-dessus, nous avons observé comment le filtrage basé sur un champ partitionné peut améliorer la récupération des données. Cependant, il existe des scénarios où les données ne peuvent pas être efficacement partitionnées par le champ spécifique que nous souhaitons filtrer. De plus, dans certains cas, un filtrage est requis sur plusieurs champs. Cela signifie que tous les fichiers d’entrée seront ouverts, ce qui peut nuire aux performances.

Heureusement, Parquet propose une solution intelligente pour atténuer ce problème. Les fichiers Parquet sont divisés en groupes de lignes. Dans chaque groupe de lignes, Parquet stocke des métadonnées. Ces métadonnées incluent les valeurs minimales et maximales pour chaque champ.

Lors de l’écriture de fichiers Parquet avec Pandas, vous pouvez sélectionner le nombre de records dans chaque groupe de contrôle.

Lors de l’utilisation de Pandas pour lire des fichiers Parquet avec des filtres, la bibliothèque Pandas exploite ces métadonnées Parquet pour filtrer efficacement les données chargées en mémoire. Si le champ souhaité se situe en dehors de la plage min/max d’un groupe de lignes, ce groupe de lignes entier est gracieusement ignoré.

df = pd.DataFrame(data)

# écriture de données non partitionnées, spécifiant la taille du groupe de lignes
df.to_parquet(path=PATH_TO_PARQUET_FILE, row_group_size=1000000)

# lecture de données non partitionnées et filtrage par groupes de lignes uniquement
df = pd.read_parquet(path=DATASET_PATH, filters=[('state','==','California')])

Voir les métadonnées à l’intérieur des fichiers Parquet peut être fait en utilisant PyArrow.

>>> import pyarrow.parquet as pq

>>> parquet_file = pq.ParquetFile(PATH_TO_PARQUET_FILE)

>>> parquet_file.metadata
<pyarrow._parquet.FileMetaData object at 0x125b21220>
  created_by: parquet-cpp-arrow version 11.0.0
  num_columns: 6
  num_rows: 1000000
  num_row_groups: 10
  format_version: 2.6
  serialized_size: 9325

>>> parquet_file.metadata.row_group(0).column(3)
<pyarrow._parquet.ColumnChunkMetaData object at 0x125b5b180>
  file_offset: 1675616
  file_path: 
  physical_type: BYTE_ARRAY
  num_values: 100000
  path_in_schema: state
  is_stats_set: True
  statistics:
    <pyarrow._parquet.Statistics object at 0x115283590>
      has_min_max: True
      min: Alabama
      max: Wyoming
      null_count: 0
      distinct_count: 0
      num_values: 100000
      physical_type: BYTE_ARRAY
      logical_type: String
      converted_type (legacy): UTF8
  compression: SNAPPY
  encodings: ('RLE_DICTIONARY', 'PLAIN', 'RLE')
  has_dictionary_page: True
  dictionary_page_offset: 1599792
  data_page_offset: 1600354
  total_compressed_size: 75824
  total_uncompressed_size: 75891

Notez que le nombre de groupes de lignes est mentionné dans les métadonnées de l’ensemble du fichier, et les valeurs minimales et maximales sont mentionnées dans la section des statistiques de chaque colonne pour chaque groupe de lignes.

Cependant, il existe une méthode pour exploiter encore davantage cette fonctionnalité Parquet pour des résultats encore plus optimisés : tri.

Filtrage par Champs Triés

Comme mentionné dans la section précédente, une partie des métadonnées stockées par Parquet inclut les valeurs minimales et maximales pour chaque champ dans chaque groupe de lignes. Lorsque les données sont triées en fonction du champ par lequel nous voulons filtrer, Pandas a une plus grande probabilité d’ignorer plus de groupes de lignes.

Par exemple, considérons un ensemble de données qui inclut une liste de dossiers, dont l’un des champs représente le ‘état’. Si les dossiers ne sont pas triés, il y a de fortes chances que chaque état apparaisse dans la plupart des groupes de lignes. Par exemple, regardez les métadonnées de la section précédente. Vous pouvez voir que le 1er groupe de lignes contient seul tous les états de ‘Alabama’ à ‘Wyoming’.

Cependant, si nous trions les données en fonction du champ ‘état’, il y a une forte probabilité de sauter de nombreux groupes de lignes.

df = pd.DataFrame(data)

# tri des données en fonction de 'état'
df.sort_values("state").to_parquet(path=NON_PARTITIONED_SORTED_PATH)

Maintenant, reprenons les métadonnées et voyons comment elles ont changé.

>>> parquet_file = pq.ParquetFile(PATH_TO_PARQUET_FILE)

>>> parquet_file.metadata.row_group(0).column(3).statistics.min
'Alabama'
>>> parquet_file.metadata.row_group(0).column(3).statistics.max
'Kentucky'


>>> parquet_file.metadata.row_group(1).column(3).statistics.min
'Kentucky'
>>> parquet_file.metadata.row_group(1).column(3).statistics.max
'North Dakota'


>>> parquet_file.metadata.row_group(2).column(3).statistics.min
'North Dakota'
>>> parquet_file.metadata.row_group(2).column(3).statistics.max
'Wyoming'

Comme vous pouvez le constater, après avoir trié par état, les valeurs min-max sont affectées en conséquence; chaque groupe de lignes contient une partie des états au lieu de tous les états. Cela signifie que la lecture avec des filtres devrait être beaucoup plus rapide maintenant.

Voyons maintenant comment cela affecte la performance de la lecture des données. Le code pour lire les données n’a pas changé.

# lecture de données non partitionnées et filtrage par groupes de lignes, les données d'entrée sont triées par état
start_time = time.time()
df = pd.read_parquet(path=DATASET_PATH, filters=[('state','==','California')])
runtime = (time.time()) - start_time # 0.24 seconds

Étonnamment, la performance ici est presque aussi bonne que l’utilisation de partitions.

Ce principe s’applique à la fois aux données partitionnées et non partitionnées. Nous pouvons utiliser les deux méthodes en même temps. Si nous voulons parfois filtrer les données en fonction du champ A et d’autres fois en fonction du champ B, alors la partition par le champ A et le tri par le champ B pourrait être une bonne option.
Dans d’autres cas, par exemple, lorsque le champ par lequel nous voulons filtrer est un champ à haute cardinalité, nous pourrions partitionner par une certaine fonction de hachage de la valeur (étiquetage) et trier les données à l’intérieur par la valeur réelle du champ de cette manière, nous bénéficierons des avantages des deux méthodes – la partition et les groupes de lignes.

Lecture d’un sous-ensemble des colonnes

Bien que moins couramment utilisé, une autre méthode pour obtenir de meilleurs résultats lors de la récupération des données consiste à sélectionner uniquement les champs spécifiques qui sont essentiels pour votre tâche. Cette stratégie peut parfois améliorer les performances. Cela est dû à la nature du format Parquet. Parquet est implémenté sous forme de format à colonnes, ce qui signifie qu’il stocke les données colonne par colonne à l’intérieur de chaque groupe de lignes. La lecture seulement de certaines colonnes signifie que les autres colonnes seront ignorées.

start_time = time.time()
df = pd.read_parquet(path=NON_PARTITIONED_SORTED_PATH, columns=["name", "state"])
runtime = (time.time()) - start_time # 0.08 seconds

Sans surprise, l’amélioration des performances est importante.

Conclusion

Bien que la partition des données soit généralement la meilleure approche, ce n’est pas toujours possible. Le tri des données peut conduire à des améliorations significatives. Nous pouvons sauter plus de groupes de lignes. De plus, si possible, la sélection seulement des colonnes nécessaires est toujours une bonne option.

Cet article vous a aidé à comprendre comment exploiter le pouvoir de parquet et pandas pour une meilleure performance.

Voici un script contenant tous les exemples mentionnés précédemment, complets avec des comparaisons de temps.