### CSV 转 Parquet 的 Python 实现及操作指南 #### 1. 使用 Pandas 将 CSV 转换为 Parquet 格式 Pandas 提供了便捷的接口用于将 CSV 文件转换为 Parquet 格式。以下是一个完整的示例： ```python import pandas as pd # 指定 CSV 文件路径 csv_path = 'data.csv' # 读取 CSV 文件为 DataFrame df = pd.read_csv(csv_path) # 指定输出的 Parquet 文件路径 parquet_path = 'data.parquet' # 将 DataFrame 写入 Parquet 文件 df.to_parquet(parquet_path) # 验证是否成功写入 read_df = pd.read_parquet(parquet_path) print(read_df) ``` 该方法适用于小型数据集，Pandas 会将整个数据集加载到内存中进行处理，因此不适用于超大规模数据。 #### 2. 使用 PySpark 将 CSV 转换为 Parquet 格式 对于大规模数据集，推荐使用 PySpark 进行分布式处理。以下是一个示例： ```python from pyspark.sql import SparkSession # 创建 SparkSession spark = SparkSession.builder.appName("CSV2Parquet").getOrCreate() # 读取 CSV 文件 df = spark.read.csv('data.csv', header=True, inferSchema=True) # 写入 Parquet 文件 df.write.parquet('data-parquet') # 读取 Parquet 文件并显示 parquet_df = spark.read.parquet('data-parquet') parquet_df.show() # 停止 SparkSession spark.stop() ``` 使用 Spark 可以处理 PB 级别的数据，适合在 Hadoop 或其他分布式环境中使用[^3]。 #### 3. 使用 PyArrow 将 CSV 转换为 Parquet 格式 PyArrow 是 Apache Arrow 的 Python 接口，提供高效的内存数据交换能力，适用于高性能数据处理： ```python import pyarrow as pa import pyarrow.parquet as pq import pandas as pd # 读取 CSV 文件为 Pandas DataFrame df = pd.read_csv('data.csv') # 转换为 Arrow Table table = pa.Table.from_pandas(df) # 写入 Parquet 文件 pq.write_table(table, 'data.parquet') # 读取 Parquet 文件并转换为 Pandas DataFrame read_table = pq.read_table('data.parquet') read_df = read_table.to_pandas() print(read_df) ``` PyArrow 在性能和内存效率上优于 Pandas，尤其适用于需要高性能数据处理的场景[^1]。 --- ### Parquet 文件的增删改查操作 #### 查询数据 使用 Pandas 读取 Parquet 文件并进行查询： ```python import pandas as pd # 读取 Parquet 文件 df = pd.read_parquet('data.parquet') # 查询特定列 print(df[['id', 'name']]) # 条件查询 print(df[df['age'] > 30]) ``` 使用 PySpark 查询 Parquet 文件： ```python from pyspark.sql import SparkSession spark = SparkSession.builder.appName("ParquetQuery").getOrCreate() df = spark.read.parquet("data-parquet") df.createOrReplaceTempView("data") # 使用 SQL 查询 result = spark.sql("SELECT * FROM data WHERE age > 30") result.show() ``` #### 更新数据 Parquet 文件本身是不可变的，无法直接进行更新操作。通常的做法是读取原始数据，修改后重新写入新的 Parquet 文件： ```python import pandas as pd # 读取 Parquet 文件 df = pd.read_parquet('data.parquet') # 修改数据 df.loc[df['id'] == 1, 'name'] = 'new_name' # 写入新文件 df.to_parquet('data_updated.parquet') ``` #### 删除数据 同样，Parquet 文件不支持直接删除操作，需通过过滤后写入新文件实现： ```python import pandas as pd # 读取 Parquet 文件 df = pd.read_parquet('data.parquet') # 删除 id 为 1 的记录 df = df[df['id'] != 1] # 写入新文件 df.to_parquet('data_filtered.parquet') ``` #### 添加数据 添加新数据时，需将原始数据与新数据合并后写入新文件： ```python import pandas as pd # 读取原始数据 df = pd.read_parquet('data.parquet') # 创建新数据 new_data = pd.DataFrame({ 'id': [5], 'name': ['new_user'], 'age': [25] }) # 合并数据 combined_df = pd.concat([df, new_data], ignore_index=True) # 写入新文件 combined_df.to_parquet('data_combined.parquet') ``` --- ###