在大数据处理中，Parquet 是一种非常常用的列式存储格式，具有压缩率高、读取效率高等优点。将多个 CSV 文件按分区存储为 Parquet 格式，可以使用 Python 的 `pandas` 和 `pyarrow` 库来实现。 以下是一个完整的示例，展示如何将多个 CSV 文件按某个字段（如日期）进行分区并保存为 Parquet 格式。 --- ### ✅ 示例代码 ```python import pandas as pd import os import pyarrow.parquet as pq import pyarrow as pa # 1. 定义CSV文件路径和输出目录 csv_files = ["data_202301.csv", "data_202302.csv", "data_202303.csv"] output_dir = "parquet_output" # 假设每个CSV文件中有一个'date'字段用于分区 def csv_to_partitioned_parquet(csv_files, output_dir, partition_col): # 创建输出目录 if not os.path.exists(output_dir): os.makedirs(output_dir) # 遍历每个CSV文件 for file in csv_files: df = pd.read_csv(file) # 确保分区字段存在 if partition_col not in df.columns: raise ValueError(f"Column '{partition_col}' not found in {file}") # 按照分区字段分组 for group_name, group_data in df.groupby(partition_col): # 构造子目录路径 partition_path = os.path.join(output_dir, f"{partition_col}={group_name}") # 创建子目录（如果不存在） if not os.path.exists(partition_path): os.makedirs(partition_path) # 构造Parquet文件名 parquet_file = os.path.join(partition_path, f"{os.path.splitext(os.path.basename(file))[0]}.parquet") # 将DataFrame转换为Parquet格式 table = pa.Table.from_pandas(group_data) pq.write_table(table, parquet_file) print(f"Saved {parquet_file}") # 调用函数 csv_to_partitioned_parquet(csv_files, output_dir, 'date') ``` --- ### ✅ 解释 1. **依赖库**： - `pandas`：用于读取 CSV 文件并进行数据处理。 - `pyarrow`：用于将 Pandas DataFrame 转换为 Arrow Table。 - `pyarrow.parquet`：用于写入 Parquet 文件。 2. **流程说明**： - 遍历所有 CSV 文件； - 对每个 CSV 按指定列（如 `date`）进行分组； - 每个分组对应一个分区目录（如 `date=202301`）； - 在每个分区目录下写入对应的 Parquet 文件。 3. **输出结构示例**： ``` parquet_output/ ├── date=202301/ │ └── data_202301.parquet ├── date=202302/ │ └── data_202302.parquet └── date=202303/ └── data_202303.parquet ``` --- ###