如何使用pyarrow從S3讀取地板文件列表作爲熊貓數據框？

Question

我使用boto3（1.4.4），pyarrow（0.4.1）和pandas（0.20.3）來做到這一點。

首先，我可以在本地讀一個單一的實木複合地板的文件是這樣的：

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's documentation後，這似乎並不可能at the moment。所以，我想出了以下解決方案：

讀從S3的單個文件，並得到一個大熊貓數據幀：

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() 

在這裏，我的哈克，不那麼優化的解決方案來建立從熊貓數據幀S3文件夾路徑：

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

您可以使用dask中的s3fs實現s3的文件系統接口。然後你可以使用ParquetDataset的文件系統參數，像這樣：

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

Answer 2

提議yjk21您應該使用s3fs模塊。然而，由於調用ParquetDataset，你會得到一個pyarrow.parquet.ParquetDataset對象。要獲得大熊貓數據幀，你寧可想要應用.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()