我正在寻找使用python从s3读取多个分区目录数据的方法.
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的ParquetDataset模块具有从分区读取的能力.所以我尝试了以下代码:
>>> import pandas as pd>>> import pyarrow.parquet as pq>>> import s3fs>>> a = "s3://my_bucker/path/to/data_folder/">>> dataset = pq.ParquetDataset(a)
它引发了以下错误:
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/
根据pyarrow的文档,我尝试使用s3fs作为文件系统,即:
>>> dataset = pq.ParquetDataset(a,filesystem=s3fs)
这会引发以下错误:
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'
我只能使用ECS集群,因此不能选择spark / pyspark.
有没有办法我们可以轻松地从s3中的这些分区目录中的python中轻松读取镶木地板文件?我认为列出所有目录,然后阅读这不是一个很好的做法,如link所示.我需要将读取数据转换为pandas数据帧以进行进一步处理&因此更喜欢与fastparquet或pyarrow相关的选项.我也对python中的其他选项持开放态度.
解决方法:
我设法使用最新版本的fastparquet& s3fs.以下是相同的代码:
import s3fsimport fastparquet as fps3 = s3fs.S3FileSystem()fs = s3fs.core.S3FileSystem()#mybucket/data_folder/serial_number=1/cur_date=20-12-2012/abcdsd0324324.snappy.parquet s3_path = "mybucket/data_folder/*/*/*.parquet"all_paths_from_s3 = fs.glob(path=s3_path)myopen = s3.open#use s3fs as the filesystemfp_obj = fp.ParquetFile(all_paths_from_s3,open_with=myopen)#convert to pandas dataframedf = fp_obj.to_pandas()
通过我们的conversation向马丁指出我正确的方向
注意:根据benchmark,这比使用pyarrow要慢.一旦通过ARROW-1213在pyarrow中实现s3fs支持,我将更新我的答案
我使用pyarrow& amp;进行了单独迭代的快速基准测试.文件列表以glob形式发送到fastparquet.对于s3fs vs pyarrow,fastparquet更快,我的hackish代码.但我认为pyarrow s3fs一旦实现就会更快.
代码&基准如下:
>>> def test_pq():... for current_file in list_parquet_files:... f = fs.open(current_file)... df = pq.read_table(f).to_pandas()... # following code is to extract the serial_number & cur_date values so that we can add them to the dataframe... #probably not the best way to split :)... elements_list=current_file.split('/')... for item in elements_list:... if item.find(date_partition) != -1:... current_date = item.split('=')[1]... elif item.find(dma_partition) != -1:... current_dma = item.split('=')[1]... df['serial_number'] = current_dma... df['cur_date'] = current_date... list_.append(df)... frame = pd.concat(list_)...>>> timeit.timeit('test_pq()',number =10,globals=globals())12.078817503992468>>> def test_fp():... fp_obj = fp.ParquetFile(all_paths_from_s3,open_with=myopen)... df = fp_obj.to_pandas()>>> timeit.timeit('test_fp()',number =10,globals=globals())2.961556333000317
更新2019年
在所有PR之后,问题如Arrow-2038& Fast Parquet – PR#182已经解决.
使用Pyarrow读取镶木地板文件
# pip install pyarrow# pip install s3fs>>> import s3fs>>> import pyarrow.parquet as pq>>> fs = s3fs.S3FileSystem()>>> bucket = 'your-bucket-name'>>> path = 'directory_name' #if its a directory omit the traling />>> bucket_uri = f's3://{bucket}/{path}''s3://your-bucket-name/directory_name'>>> dataset = pq.ParquetDataset(bucket_uri, filesystem=fs)>>> table = dataset.read()>>> df = table.to_pandas()
使用Fast parquet阅读镶木地板文件
# pip install s3fs# pip install fastparquet>>> import s3fs>>> import fastparquet as fp>>> bucket = 'your-bucket-name'>>> path = 'directory_name'>>> root_dir_path = f'{bucket}/{path}'# the first two wild card represents the 1st,2nd column partitions columns of your data & so forth>>> s3_path = f"{root_dir_path}/*/*/*.parquet">>> all_paths_from_s3 = fs.glob(path=s3_path)>>> fp_obj = fp.ParquetFile(all_paths_from_s3,open_with=myopen, root=root_dir_path)>>> df = fp_obj.to_pandas()
快速基准
这可能不是对其进行基准测试的最佳方式.请阅读blog post以获得直通基准
#pyarrow>>> import timeit>>> def test_pq():... dataset = pq.ParquetDataset(bucket_uri, filesystem=fs)... table = dataset.read()... df = table.to_pandas()...>>> timeit.timeit('test_pq()',number =10,globals=globals())1.2677053569998407#fastparquet>>> def test_fp():... fp_obj = fp.ParquetFile(all_paths_from_s3,open_with=myopen, root=root_dir_path)... df = fp_obj.to_pandas()>>> timeit.timeit('test_fp()',number =10,globals=globals())2.931876824000028
关于Pyarrow的speed的进一步阅读
参考:
> fastparquet
> s3fs
> pyarrow
>基于discussion& ;;的pyarrow箭头代码还有文件
>基于讨论的fastparquet代码PR-182,PR-182&还有文件
联系客服