使用包pyarrow和pandas，您可以将CSV转换为Parquet，而无需在后台使用JVM：

import pandas as pd
df = pd.read_csv('example.csv')
df.to_parquet('output.parquet')

运行pyarrow的一个限制是，pyarrow只在Windows上的Python 3.5+中可用，要么使用Linux/OSX作为Python 2运行代码，要么将Windows安装升级到Python 3.6。

你可以只使用pyarrow将csv转换成parquet格式--不使用panda。当你需要最小化代码依赖性时，它可能会很有用（例如，与AWS Lambda）。

import pyarrow.csv as pv
import pyarrow.parquet as pq

table = pv.read_csv(filename)
pq.write_table(table, filename.replace('csv', 'parquet'))

请参阅pyarrow文档以微调read_csv和write_table函数。

import boto3
import pandas as pd
import pyarrow as pa
from s3fs import S3FileSystem
import pyarrow.parquet as pq

s3 = boto3.client('s3',region_name='us-east-2')
obj = s3.get_object(Bucket='ssiworkoutput', Key='file_Folder/File_Name.csv')
df = pd.read_csv(obj['Body'])

table = pa.Table.from_pandas(df)

output_file = "s3://ssiworkoutput/file/output.parquet"  # S3 Path need to mention
s3 = S3FileSystem()

pq.write_to_dataset(table=table,
                    root_path=output_file,partition_cols=['Year','Month'],
                    filesystem=s3)

print("File converted from CSV to parquet completed")

有几种不同的方法可以用Python将CSV文件转换为Parquet格式。
Uwe L. Korn的Pandas方法非常有效。
如果您想将多个CSV文件转换为多个Parquet /单个Parquet文件，请使用Dask。这会将多个CSV文件转换为两个Parquet文件：

import dask.dataframe as dd

df = dd.read_csv('./data/people/*.csv')
df = df.repartition(npartitions=4)
df.to_parquet('./tmp/people_parquet4')

如果你只想输出一个Parquet文件，你也可以使用df.repartition(npartitions=1)。更多关于使用Dask here将CSV转换为Parquet的信息。
下面是一个在Spark环境中工作的PySpark代码片段：

from pyspark.sql import SparkSession

spark = SparkSession.builder \
  .master("local") \
  .appName("parquet_example") \
  .getOrCreate()

df = spark.read.csv('data/us_presidents.csv', header = True)
df.repartition(1).write.mode('overwrite').parquet('tmp/pyspark_us_presidents')

您也可以在Spark环境中使用Koalas：

import databricks.koalas as ks

df = ks.read_csv('data/us_presidents.csv')
df.to_parquet('tmp/koala_us_presidents')

处理大于内存的CSV文件

下面的代码将CSV转换为Parquet，而无需将整个CSV文件加载到内存中

import pandas as pd
import pyarrow as pa
import pyarrow.parquet as pq

new_schema = pa.schema([
    ('col1', pa.int64()),
    ('col2', pa.int64()),
    ('newcol', pa.int64())
])

csv_column_list = ['col1', 'col2']

with pq.ParquetWriter('my_parq_data.parquet', schema=new_schema) as writer:
    with pd.read_csv('my_data.csv', header=None, names=csv_column_list, chunksize=100000) as reader:
        for df in reader:
            # transformation: transform df by adding a new static column with column name 'newcol' and value 9999999
            df['newcol'] = 9999999
            # convert pandas df to record batch
            transformed_batch = pa.RecordBatch.from_pandas(df, schema=new_schema)
            writer.write_batch(transformed_batch)

以上代码：
1.读取大型CSV file in chunks。
1.通过添加新列转换数据框。
1.将df转换为箭头记录批。
1.将已转换的箭头批次作为新行组写入 parquet 文件。

注意：不要将块大小保持得太小。这将导致压缩效果不佳，因为块大小也与新 parquet 文件中的行组大小相对应。

您可以使用spark写入为PARQUET FILE：

spark = SparkSession.builder.appName("Test_Parquet").master("local[*]").getOrCreate()

parquetDF = spark.read.csv("data.csv")

parquetDF.coalesce(1).write.mode("overwrite").parquet("Parquet")

我希望这能帮上忙

from pyspark import SparkContext
from pyspark.sql import SQLContext
from pyspark.sql.types import *
import sys

sc = SparkContext(appName="CSV2Parquet")
sqlContext = SQLContext(sc)

schema = StructType([
    StructField("col1", StringType(), True),
    StructField("col2", StringType(), True),
    StructField("col3", StringType(), True),
    StructField("col4", StringType(), True),
    StructField("col5", StringType(), True)])
rdd = sc.textFile('/input.csv').map(lambda line: line.split(","))
df = sqlContext.createDataFrame(rdd, schema)
df.write.parquet('/output.parquet')