J'écris un fichier parquet à partir d'un Spark DataFrame de la manière suivante:
import pandas as pd
df = pd.read_parquet("path/myfile.parquet", engine="pyarrow")
Cela crée un dossier contenant plusieurs fichiers.
Quand j'essaye de lire ceci dans les pandas, j'obtiens les erreurs suivantes, selon l'analyseur que j'utilise:
df.write.parquet("path/myfile.parquet", mode = "overwrite", compression="gzip")
PyArrow:
Fichier "pyarrow \ error.pxi", ligne 83, dans pyarrow.lib.check_status
ArrowIOError: Fichier parquet non valide. Pied de page corrompu.
Fichier "C: \ Program Files \ Anaconda3 \ lib \ site-packages \ fastparquet \ util.py", ligne 38, dans default_open retourne ouvert (f, mode)
PermissionError: [Errno 13] Permission refusée: 'path / myfile.parquet'
J'utilise les versions suivantes:
J'ai essayé gzip ainsi que la compression rapide. Les deux ne fonctionnent pas. Bien sûr, je me suis assuré d'avoir le fichier dans un emplacement où Python a les autorisations de lecture / écriture.
Cela aiderait déjà si quelqu'un était capable de reproduire cette erreur.
3 Réponses :
Le problème est que Spark partitionne le fichier en raison de sa nature distribuée (chaque exécuteur écrit un fichier dans le répertoire qui reçoit le nom de fichier). Ce n'est pas quelque chose pris en charge par Pandas, qui attend un fichier, pas un chemin.
Vous pouvez contourner ce problème de différentes manières:
Lire le fichier avec un utilitaire alternatif, tel que pyarrow.parquet.ParquetDataset , puis le convertir en Pandas (je n'ai pas testé ce code).
arrow_dataset = pyarrow.parquet.ParquetDataset('path/myfile.parquet')
arrow_table = arrow_dataset.read()
pandas_df = arrow_table.to_pandas()
Une autre façon est de lire les fragments séparés séparément, puis de les concaténer, comme le suggère cette réponse: Lire plusieurs fichiers parquet dans un dossier et écrire dans un seul fichier csv en utilisant python
Merci pour votre réponse. Il semble que la lecture de fichiers uniques (votre deuxième puce) fonctionne. Cependant, la première chose ne fonctionne pas - il semble que pyarrow ne puisse pas gérer le pied de page de PySpark (voir le message d'erreur en question)
@Thomas, je ne suis malheureusement pas sûr du problème du pied de page.
Ou vous pouvez essayer d'appeler coalesce sur la trame de données: coalesce (1) afin de fusionner tous les fichiers de pièce en un seul fichier, puis de lire à partir du fichier unique au lieu d'un répertoire de fichiers?
@OmkarNeogi: Ceci n'est possible que si vous êtes la personne qui écrit les fichiers, pas si vous les recevez de quelqu'un d'autre ...
J'ai mis à jour cela pour travailler avec les API réelles, c'est-à-dire que vous créez un ensemble de données, le convertissez en table, puis en un Pandas DataFrame.
Comme cela semble toujours être un problème même avec les nouvelles versions de pandas, j'ai écrit quelques fonctions pour contourner cela dans le cadre d'une plus grande bibliothèque de helpers pyspark:
import pandas as pd
import datetime
def read_parquet_folder_as_pandas(path, verbosity=1):
files = [f for f in os.listdir(path) if f.endswith("parquet")]
if verbosity > 0:
print("{} parquet files found. Beginning reading...".format(len(files)), end="")
start = datetime.datetime.now()
df_list = [pd.read_parquet(os.path.join(path, f)) for f in files]
df = pd.concat(df_list, ignore_index=True)
if verbosity > 0:
end = datetime.datetime.now()
print(" Finished. Took {}".format(end-start))
return df
def read_parquet_as_pandas(path, verbosity=1):
"""Workaround for pandas not being able to read folder-style parquet files.
"""
if os.path.isdir(path):
if verbosity>1: print("Parquet file is actually folder.")
return read_parquet_folder_as_pandas(path, verbosity)
else:
return pd.read_parquet(path)
Cela suppose que les fichiers pertinents dans le "dossier" parquet, qui est en fait un dossier, se terminent par ".parquet". Cela fonctionne pour les fichiers de parquet exportés par databricks et peut également fonctionner avec d'autres (non testé, satisfait des commentaires dans les commentaires).
La fonction read_parquet_as_pandas () peut être utilisée si elle on ne sait pas à l'avance s'il s'agit d'un dossier ou non.
Si le fichier parquet a été créé avec spark, (donc c'est un répertoire) pour l'importer vers les pandas utilisez
from pyarrow.parquet import ParquetDataset
dataset = ParquetDataset("file.parquet")
table = dataset.read()
df = table.to_pandas()