使用Python将CSV文件转换为Parquet文件
使用Python将CSV文件转换为Parquet文件
我正在尝试将一个.csv文件转换为.parquet文件。
csv文件（Temp.csv）的格式如下：
1,Jon,Doe,Denver
我正在使用以下python代码将其转换为parquet：
from pyspark import SparkContext
from pyspark.sql import SQLContext
from pyspark.sql.types import *
import os
if __name__ == "__main__":
sc = SparkContext(appName="CSV2Parquet")
sqlContext = SQLContext(sc)
schema = StructType([
StructField("col1", IntegerType(), True),
StructField("col2", StringType(), True),
StructField("col3", StringType(), True),
StructField("col4", StringType(), True)])
dirname = os.path.dirname(os.path.abspath(__file__))
csvfilename = os.path.join(dirname,'Temp.csv')
rdd = sc.textFile(csvfilename).map(lambda line: line.split(","))
df = sqlContext.createDataFrame(rdd, schema)
parquetfilename = os.path.join(dirname,'output.parquet')
df.write.mode('overwrite').parquet(parquetfilename)
结果只是一个名为output.parquet的文件夹，而不是我要找的parquet文件，并且控制台上出现以下错误。
[CSV to Parquet Error](https://i.stack.imgur.com/WWDID.png)
我还尝试运行以下代码，遇到了类似的问题。
from pyspark.sql import SparkSession
import os
spark = SparkSession \
.builder \
.appName("Protob Conversion to Parquet") \
.config("spark.some.config.option", "some-value") \
.getOrCreate()
# 读取csv文件
dirname = os.path.dirname(os.path.abspath(__file__))
csvfilename = os.path.join(dirname,'Temp.csv')
df = spark.read.csv(csvfilename)
# 将DataFrame的内容显示到stdout
df.show()
parquetfilename = os.path.join(dirname,'output.parquet')
df.write.mode('overwrite').parquet(parquetfilename)
如何最好地做到这一点？使用Windows和Python 2.7。
问题的出现原因：需要将CSV文件转换为Parquet文件，并且在转换过程中进行分区。
解决方法：使用Python编写代码实现转换和分区的功能。
文章内容如下：
使用Python将CSV文件转换为Parquet文件
在数据处理过程中，经常需要将CSV文件转换为Parquet文件来提高数据处理的效率和性能。本文将介绍如何使用Python将CSV文件转换为Parquet文件，并在转换过程中进行分区。
首先，我们需要安装一些必要的库。在运行代码之前，请确保已安装以下库：
pip3 install boto3
pip3 install pandas
pip3 install pyarrow
pip3 install fs-s3fs
pip3 install s3fs
接下来，我们可以使用以下代码将CSV文件从AWS S3路径读取，并将其存储为Parquet格式，并在转换过程中进行分区：
import boto3
import pandas as pd
import pyarrow as pa
from s3fs import S3FileSystem
import pyarrow.parquet as pq
# 创建S3客户端
s3 = boto3.client('s3',region_name='us-east-2')
# 从S3路径获取CSV文件对象
obj = s3.get_object(Bucket='ssiworkoutput', Key='file_Folder/File_Name.csv')
# 读取CSV文件为Pandas DataFrame
df = pd.read_csv(obj['Body'])
# 将DataFrame转换为PyArrow Table
table = pa.Table.from_pandas(df)
# 指定输出文件路径
output_file = "s3://ssiworkoutput/file/output.parquet"
# 创建S3文件系统对象
s3 = S3FileSystem()
# 将Table以Parquet格式写入S3路径，并进行分区
pq.write_to_dataset(table=table,
root_path=output_file,
partition_cols=['Year','Month'],
filesystem=s3)
print("文件从CSV转换为Parquet格式完成")
运行以上代码，即可将CSV文件转换为Parquet文件，并在转换过程中进行分区。转换后的Parquet文件将存储在指定的S3路径中。
如何在AWS上安装pyarrow而不会遇到包大小问题？
如果在AWS上安装pyarrow时遇到包大小问题，最简单的方法是使用awswrangler层，该层已经包含了pyarrow库。
希望本文能对您在使用Python将CSV文件转换为Parquet文件时有所帮助。
使用pyarrow和pandas包，可以在不使用后台JVM的情况下将CSV文件转换为Parquet文件：
import pandas as pd
df = pd.read_csv('example.csv')
df.to_parquet('output.parquet')
一个限制是pyarrow只适用于Windows上的Python 3.5+。要么在Linux/OSX上运行代码作为Python 2，要么将Windows设置升级为Python 3.6。
谢谢你的回答。在Windows上没有使用Python 2.7的方法吗？
这是将单个文件转换为parquet文件的非常简单的方法，但是如果我们有多个csv文件并且想要将它们合并为一个parquet文件，应该怎么办？
您可以遍历文件并将每个文件转换为parquet，如果您希望使用Python之外的其他方法，AWS EMR上的Hive在将csv转换为parquet方面效果很好。
您可以使用Dask或PySpark将多个CSV文件转换为单个Parquet文件（或多个Parquet文件）。有关更多细节，请参阅我的答案。
在某些情况下，我们可能需要将CSV文件转换为Parquet文件，而不依赖于Pandas库，而是仅使用pyarrow库来实现。这种情况可能出现在我们需要减少代码依赖性的情况下，例如在AWS Lambda上运行代码时。
为了实现CSV到Parquet的转换，我们可以使用pyarrow库中的`pyarrow.csv`和`pyarrow.parquet`模块。下面是一个示例代码：
import pyarrow.csv as pv
import pyarrow.parquet as pq
# 读取CSV文件
table = pv.read_csv(filename)
# 将数据写入Parquet文件
pq.write_table(table, filename.replace('csv', 'parquet'))
在上述代码中，我们首先使用`pv.read_csv`函数读取CSV文件并将其转换为pyarrow的Table对象。然后，我们使用`pq.write_table`函数将Table对象写入Parquet文件。需要注意的是，我们在写入文件时将文件扩展名从"csv"替换为"parquet"。
如果需要进一步调整转换过程的设置，可以参考pyarrow文档中的`read_csv`和`write_table`函数。
通过这种方式，我们可以使用pyarrow库将CSV文件转换为Parquet文件，而无需依赖于Pandas库。这在需要减少代码依赖性的情况下非常有用，特别是在使用AWS Lambda等服务时。