How to read partitioned parquet files from S3 using pyarrow in python

user_1552998 0
Viewed 76119

I looking for ways to read data from multiple partitioned directories from s3 using python.

data_folder/serial_number=1/cur_date=20-12-2012/abcdsd0324324.snappy.parquet data_folder/serial_number=2/cur_date=27-12-2012/asdsdfsd0324324.snappy.parquet

pyarrow's ParquetDataset module has the capabilty to read from partitions. So I have tried the following code : 

>>> import pandas as pd
>>> import pyarrow.parquet as pq
>>> import s3fs
>>> a = "s3://my_bucker/path/to/data_folder/"
>>> dataset = pq.ParquetDataset(a)

It threw the following error :

Traceback (most recent call last):
  File "<stdin>", line 1, in <module>
  File "/home/my_username/anaconda3/lib/python3.6/site-packages/pyarrow/parquet.py", line 502, in __init__
    self.metadata_path) = _make_manifest(path_or_paths, self.fs)
  File "/home/my_username/anaconda3/lib/python3.6/site-packages/pyarrow/parquet.py", line 601, in _make_manifest
    .format(path))
OSError: Passed non-file path: s3://my_bucker/path/to/data_folder/

Based on documentation of pyarrow I tried using s3fs as the file system, ie :

>>> dataset = pq.ParquetDataset(a,filesystem=s3fs)

Which throws the following error : 

Traceback (most recent call last):
  File "<stdin>", line 1, in <module>
  File "/home/my_username/anaconda3/lib/python3.6/site-packages/pyarrow/parquet.py", line 502, in __init__
    self.metadata_path) = _make_manifest(path_or_paths, self.fs)
  File "/home/my_username/anaconda3/lib/python3.6/site-packages/pyarrow/parquet.py", line 583, in _make_manifest
    if is_string(path_or_paths) and fs.isdir(path_or_paths):
AttributeError: module 's3fs' has no attribute 'isdir'

I am limited to use a ECS cluster, hence spark/pyspark is not an option.

Is there a way we can easily read the parquet files easily, in python from such partitioned directories in s3 ? I feel that listing the all the directories and then reading the is not a good practise as suggested in this link. I would need to convert the read data to a pandas dataframe for further processing & hence prefer options related to fastparquet or pyarrow. I am open to other options in python as well.

python-s3fs fastparquet parquet pyarrow python
5 Answers

For python 3.6+ AWS has a library called aws-data-wrangler that helps with the integration between Pandas/S3/Parquet

to install do;

pip install awswrangler

to read partitioned parquet from s3 using awswrangler 1.x.x and above, do;

import awswrangler as wr
df = wr.s3.read_parquet(path="s3://my_bucket/path/to/data_folder/", dataset=True)

By setting dataset=True awswrangler expects partitioned parquet files. It will read all the individual parquet files from your partitions below the s3 key you specify in the path.

user_17711550

For those of you who want to read in only parts of a partitioned parquet file, pyarrow accepts a list of keys as well as just the partial directory path to read in all parts of the partition. This method is especially useful for organizations who have partitioned their parquet datasets in a meaningful like for example by year or country allowing users to specify which parts of the file they need. This will reduce costs in the long run as AWS charges per byte when reading in datasets. 

# Read in user specified partitions of a partitioned parquet file 

import s3fs
import pyarrow.parquet as pq
s3 = s3fs.S3FileSystem()

keys = ['keyname/blah_blah/part-00000-cc2c2113-3985-46ac-9b50-987e9463390e-c000.snappy.parquet'\
         ,'keyname/blah_blah/part-00001-cc2c2113-3985-46ac-9b50-987e9463390e-c000.snappy.parquet'\
         ,'keyname/blah_blah/part-00002-cc2c2113-3985-46ac-9b50-987e9463390e-c000.snappy.parquet'\
         ,'keyname/blah_blah/part-00003-cc2c2113-3985-46ac-9b50-987e9463390e-c000.snappy.parquet']

bucket = 'bucket_yada_yada_yada'

# Add s3 prefix and bucket name to all keys in list
parq_list=[]
for key in keys:
    parq_list.append('s3://'+bucket+'/'+key)

# Create your dataframe
df = pq.ParquetDataset(parq_list, filesystem=s3).read_pandas(columns=['Var1','Var2','Var3']).to_pandas()
user_65151100

This issue was resolved in this pull request in 2017.

For those who want to read parquet from S3 using only pyarrow, here is an example:

import s3fs
import pyarrow.parquet as pq

fs = s3fs.S3FileSystem()
bucket = "your-bucket"
path = "your-path"

# Python 3.6 or later
p_dataset = pq.ParquetDataset(
    f"s3://{bucket}/{path}",
    filesystem=fs
)
df = p_dataset.read().to_pandas()

# Pre-python 3.6
p_dataset = pq.ParquetDataset(
    "s3://{0}/{1}".format(bucket, path),
    filesystem=fs
)
df = p_dataset.read().to_pandas()
user_42609910

PyArrow 7.0.0 has some improvements to a new module, pyarrow.dataset, that is meant to abstract away the dataset concept from the previous, Parquet-specific pyarrow.parquet.ParquetDataset.

Assuming you are fine with the dataset schema being inferred from the first file, the example from the documentation for reading a partitioned dataset should just work.

Here's a more-complete example assuming you want to use data from S3:

import pyarrow.dataset as ds
from pyarrow import fs

s3 = fs.S3FileSystem()

dataset = ds.dataset(
    "my-bucket-name/my-path-to-dataset-partitions",
    format="parquet",
    filesystem=s3,
    partitioning="hive"
)

# Assuming your data is partitioned like year=2022/month=4/day=29
# this will only have to read the files for that day

expression = ((ds.field("year") == 2022) & (ds.field("month") == 4) & (ds.field("day") == 29))

pyarrow_table_2022_04_29 = dataset.to_table(filter=expression)

A word of warning if you're defining the dataset schema yourself. That inference above with the partition argument automatically adds the partitions to your dataset schema.

If you want the partitioning to work properly with a manually-defined dataset schema, you must make sure you add the partitions to the schema:

import pyarrow as pa

my_manual_schema = pa.schema([])  # Some pyarrow.Schema instance for your dataset

# Be sure to add the partitions even though they're not in the dataset files
my_manual_schema.append(pa.field("year", pa.int16()))
my_manual_schema.append(pa.field("month", pa.int8()))
my_manual_schema.append(pa.field("day", pa.int8()))

dataset = ds.dataset(
    "my-bucket-name/my-path-to-dataset-partitions",
    format="parquet",
    filesystem=s3,
    schema=my_manual_schema,
    partitioning="hive"
)
user_68953590
Related