pythonpyarrow.parquet+pandas：读取及使用parquet文件

见贤思齐

文章目录一、前言1.所需的库2.终端指令二、pyarrow.parquet1.读取Parquet文件2.写入Parquet文件3.对数据进行操作4.导出数据为csv三、实战1.简单读取2.数据操作（分割feature）3.迭代方式来处理Parquet文件4.读取同一文件夹下多个parquet文件  Parquet是一种用于列式存储和压缩数据的文件格式，广泛应用于大数据处理和分析中。Python提供了多个库来处理Parquet文件，例如pyarrow和fastparquet。  本文将介绍如何使用pyarrow.parquet+pandas库操作Parquet文件。一、前言1.所需的库importpyarrow.parquetaspqimportpandasaspd123  pyarrow.parquet模块，可以读取和写入Parquet文件，以及进行一系列与Parquet格式相关的操作。例如，可以使用该模块读取Parquet文件中的数据，并转换为pandasDataFrame来进行进一步的分析和处理。同时，也可以使用这个模块将DataFrame的数据保存为Parquet格式。2.终端指令condacreate-nDLpython==3.111condaactivateDL1condainstallpyarrow1或pipinstallpyarrow1二、pyarrow.parquet  当使用pyarrow.parquet模块时，通常的操作包括读取和写入Parquet文件，以及对Parquet文件中的数据进行操作和转换。以下是一些常见的使用方法：1.读取Parquet文件importpyarrow.parquetaspqparquet_file=pq.ParquetFile('file.parquet')data=parquet_file.read().to_pandas()1234使用pq.ParquetFile打开Parquet文件；使用read().to_pandas()方法将文件中的数据读取为pandasDataFrame。2.写入Parquet文件importpandasaspdimportpyarrowaspaimportpyarrow.parquetaspqdf=pd.DataFrame({'col1':[1,2,3],'col2':['a','b','c']})table=pa.Table.from_pandas(df)pq.write_table(table,'output.parquet')12345678将pandasDataFrame转换为Arrow的Table格式；使用pq.write_table方法将Table写入为Parquet文件。parquet_file=pq.ParquetFile('output.parquet')data=parquet_file.read().to_pandas()print(data)12343.对数据进行操作importpyarrow.parquetaspq#读取Parquet文件parquet_file=pq.ParquetFile('output.parquet')data=parquet_file.read().to_pandas()#对数据进行筛选和转换filtered_data=data[data['col1']>1]#筛选出col1大于1的行print(filtered_data)transformed_data=filtered_data.assign(col3=filtered_data['col1']*2)#添加一个新列col3，值为col1的两倍#打印处理后的数据print(transformed_data)12345678910111213144.导出数据为csvimportpyarrow.parquetaspqimportpandasaspdparquet_file=pq.ParquetFile('output.parquet')data=parquet_file.read().to_pandas()df=pd.DataFrame(data)csv_path='./data.csv'df.to_csv(csv_path)print(f'数据已保存到{csv_path}')1234567891011三、实战1.简单读取importpyarrow.parquetaspqimportpandasaspdparquet_file=pq.ParquetFile('./train_parquet/part-00014-918feee1-1ad5-4b08-8876-4364cc996930-c000.snappy.parquet')data=parquet_file.read().to_pandas()df=pd.DataFrame(data)csv_path='./data2.csv'df.to_csv(csv_path)print(f'数据已保存到{csv_path}')1234567891011关于PyCharm调试操作可参照：PyCharm基础调试功能详解点击右侧蓝色的ViewasDataFrame  如图所示，feature在同一个格内，导出为：注意看，省略号...位置真的就是省略号字符，没有数字，即[0.270588240.0.05882353...0.478431370.368627450.97647059]12.数据操作（分割feature）importpyarrow.parquetaspqimportpandasaspdparquet_file=pq.ParquetFile('./train_parquet/part-00014-918feee1-1ad5-4b08-8876-4364cc996930-c000.snappy.parquet')data=parquet_file.read().to_pandas()#将feature列中的列表拆分成单独的特征值split_features=data['feature'].apply(lambdax:pd.Series(x))#将拆分后的特征添加到DataFrame中data=pd.concat([data,split_features],axis=1)print(data.head(2))#删除原始的feature列data=data.drop('feature',axis=1)#保存到csv文件csv_path='./data1.csv'data.to_csv(csv_path,index=False)print(f'数据已保存到{csv_path}')123456789101112131415161718192021调试打开：excel打开：文件大小对比部分内容援引自博客：使用python打开parquet文件3.迭代方式来处理Parquet文件  如果Parquet文件非常大，可能会占用大量的内存。在处理大型数据时，建议使用迭代的方式来处理Parquet文件，以减少内存的占用。以下是一种更加内存友好的方式来处理Parquet文件：importpyarrow.parquetaspqimportpandasaspdimporttimestart_time=time.time()#记录开始时间#使用迭代器迭代读取Parquet文件中的数据data_iterator=pq.ParquetFile('./train_parquet/part-00014-918feee1-1ad5-4b08-8876-4364cc996930-c000.snappy.parquet').iter_batches(batch_size=100)#初始化空的DataFrame用于存储数据data=pd.DataFrame()#逐批读取数据并进行处理forbatchindata_iterator:#将RecordBatch转换为PandasDataFramedf_batch=batch.to_pandas()#将feature列中的列表拆分成单独的特征值split_features=df_batch['feature'].apply(lambdax:pd.Series(x))#将拆分后的特征添加到DataFrame中df_batch=pd.concat([df_batch,split_features],axis=1)#将处理后的数据追加到DataFrame中data=data._append(df_batch,ignore_index=True)#删除原始的feature列data=data.drop('feature',axis=1)#保存到csv文件csv_path='./data3.csv'data.to_csv(csv_path,index=False)end_time=time.time()#记录结束时间print(f'数据已保存到{csv_path}')print(f'总运行时间:{end_time-start_time}秒')1234567891011121314151617181920212223242526272829303132333435363738输出：数据已保存到./data3.csv总运行时间:4.251184940338135秒124.读取同一文件夹下多个parquet文件importosimportpyarrow.parquetaspqimportpandasaspdimporttimestart_time=time.time()#记录开始时间folder_path='./train_parquet/'parquet_files=[fforfinos.listdir(folder_path)iff.endswith('.parquet')]#初始化空的DataFrame用于存储数据data=pd.DataFrame()#逐个读取Parquet文件中的数据并进行处理forfileinparquet_files:file_path=os.path.join(folder_path,file)data_iterator=pq.ParquetFile(file_path).iter_batches(batch_size=1024)forbatchindata_iterator:#将RecordBatch转换为PandasDataFramedf_batch=batch.to_pandas()#将feature列中的列表拆分成单独的特征值split_features=df_batch['feature'].apply(lambdax:pd.Series(x))#将拆分后的特征添加到DataFrame中df_batch=pd.concat([df_batch,split_features],axis=1)#将处理后的数据追加到DataFrame中data=data._append(df_batch,ignore_index=True)#删除原始的feature列data=data.drop('feature',axis=1)#保存到csv文件csv_path='./data.csv'data.to_csv(csv_path,index=False)end_time=time.time()#记录结束时间print(f'数据已保存到{csv_path}')print(f'总运行时间:{end_time-start_time}秒')123456789101112131415161718192021222324252627282930313233343536373839404142