如何使用pyarrow从S3读取镶木地板文件列表作为pandas数据框？

Question

我有一种使用boto3(1.4.4),pyarrow(0.4.1)和pandas(0.20.3)实现这一目标的hacky方法.

首先,我可以在本地读取单个镶木地板文件,如下所示:

import pyarrow.parquet as pq

path = 'parquet/part-r-00000-1e638be4-e31f-498a-a359-47d017a0059c.gz.parquet'
table = pq.read_table(path)
df = table.to_pandas()

我也可以在本地读取镶木地板文件目录,如下所示:

import pyarrow.parquet as pq

dataset = pq.ParquetDataset('parquet/')
table = dataset.read()
df = table.to_pandas()

两者都像魅力一样.现在我想用存储在S3存储桶中的文件远程实现相同的功能.我希望这样的东西能起作用:

dataset = pq.ParquetDataset('s3n://dsn/to/my/bucket')

但它没有:

OSError: Passed non-file path: s3n://dsn/to/my/bucket

在仔细阅读了pyarrow的文档后,目前似乎无法做到这一点.所以我提出了以下解决方案:

从S3读取单个文件并获取pandas数据帧:

import io
import boto3
import pyarrow.parquet as pq

buffer = io.BytesIO()
s3 = boto3.resource('s3')
s3_object = s3.Object('bucket-name', 'key/to/parquet/file.gz.parquet')
s3_object.download_fileobj(buffer)
table = pq.read_table(buffer)
df = table.to_pandas()

在这里,我的hacky,not-so-optimized,解决方案从S3文件夹路径创建一个pandas数据框:

import io
import boto3
import pandas as pd
import pyarrow.parquet as pq

bucket_name = 'bucket-name'
def download_s3_parquet_file(s3, bucket, key):
    buffer = io.BytesIO()
    s3.Object(bucket, key).download_fileobj(buffer)
    return buffer

client = boto3.client('s3')
s3 = boto3.resource('s3')
objects_dict = client.list_objects_v2(Bucket=bucket_name, Prefix='my/folder/prefix')
s3_keys = [item['Key'] for item in objects_dict['Contents'] if item['Key'].endswith('.parquet')]
buffers = [download_s3_parquet_file(s3, bucket_name, key) for key in s3_keys]
dfs = [pq.read_table(buffer).to_pandas() for buffer in buffers]
df = pd.concat(dfs, ignore_index=True)

有没有更好的方法来实现这一目标？也许某种使用pyarrow的熊猫连接器？我想避免使用pyspark,但如果没有其他解决方案,那么我会接受它.

Answer 1

您应该使用yjk21s3fs建议的模块.但是,如果调用ParquetDataset,您将获得一个pyarrow.parquet.ParquetDataset对象.要获得Pandas DataFrame,您宁愿申请它:.read_pandas().to_pandas()

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

pandas_dataframe = pq.ParquetDataset('s3://your-bucket/', filesystem=s3).read_pandas().to_pandas()

Answer 2

谢谢！你的问题实际上告诉了我很多。这就是我现在使用pandas(0.21.1) 执行的操作，它将调用pyarrow和boto3(1.3.1)。

import boto3
import io
import pandas as pd

# Read single parquet file from S3
def pd_read_s3_parquet(key, bucket, s3_client=None, **args):
    if s3_client is None:
        s3_client = boto3.client('s3')
    obj = s3_client.get_object(Bucket=bucket, Key=key)
    return pd.read_parquet(io.BytesIO(obj['Body'].read()), **args)

# Read multiple parquets from a folder on S3 generated by spark
def pd_read_s3_multiple_parquets(filepath, bucket, s3=None, 
                                 s3_client=None, verbose=False, **args):
    if not filepath.endswith('/'):
        filepath = filepath + '/'  # Add '/' to the end
    if s3_client is None:
        s3_client = boto3.client('s3')
    if s3 is None:
        s3 = boto3.resource('s3')
    s3_keys = [item.key for item in s3.Bucket(bucket).objects.filter(Prefix=filepath)
               if item.key.endswith('.parquet')]
    if not s3_keys:
        print('No parquet found in', bucket, filepath)
    elif verbose:
        print('Load parquets:')
        for p in s3_keys: 
            print(p)
    dfs = [pd_read_s3_parquet(key, bucket=bucket, s3_client=s3_client, **args) 
           for key in s3_keys]
    return pd.concat(dfs, ignore_index=True)

然后您可以通过以下方式从 S3 读取文件夹下的多个镶木地板

df = pd_read_s3_multiple_parquets('path/to/folder', 'my_bucket')

（我想可以大大简化这段代码。）

Answer 3

将云上的 parquet 数据读入数据帧的最简单方法可能是以这种方式使用dask.dataframe：

import dask.dataframe as dd
df = dd.read_parquet('s3://bucket/path/to/data-*.parq')

dask.dataframe可以从 Google Cloud Storage、Amazon S3、Hadoop 文件系统等读取！

Answer 4

也可以使用boto3来完成，而无需使用pyarrow

import boto3
import io
import pandas as pd

# Read the parquet file
buffer = io.BytesIO()
s3 = boto3.resource('s3')
object = s3.Object('bucket_name','key')
object.download_fileobj(buffer)
df = pd.read_parquet(buffer)

print(df.head())

Answer 5

如果您有正确的包设置

$ pip install pandas==1.1.0 pyarrow==1.0.0 s3fs==0.4.2

以及适当配置 的 AWS 共享配置和凭证文件

您可以立即使用pandas：

import pandas as pd

df = pd.read_parquet("s3://bucket/key.parquet")

如果有多个 AWS配置文件，您可能还需要设置

$ export AWS_DEFAULT_PROFILE=profile_under_which_the_bucket_is_accessible

这样您就可以访问您的存储桶。

Answer 6

您可以使用dask的s3fs，它为s3实现了文件系统接口。然后，您可以像下面这样使用ParquetDataset的文件系统参数：

import s3fs
s3 = s3fs.S3FileSystem()
dataset = pq.ParquetDataset('s3n://dsn/to/my/bucket', filesystem=s3)

Answer 7

如果您也愿意使用AWS Data Wrangler。

import awswrangler as wr

df = wr.s3.read_parquet(path="s3://...")