如何在Pandas中转换数据类型并生成Spark兼容的Parquet文件

核心思路是利用PyArrow自定义Parquet schema，强制指定列的存储类型，让Spark能正确识别timestamp和double，全程用Pandas处理，不依赖Spark DataFrame。

具体实现步骤 #

1. 依赖准备 #

Databricks 14.3 LTS默认已包含pyarrow，若环境缺失可通过%pip install pyarrow安装。

2. 代码实现 #

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

def process_my_file(netcdf_path, parquet_output_path):
    # 从netCDF读取为Pandas DataFrame
    ds = xr.open_dataset(netcdf_path)
    df = ds.to_dataframe().reset_index()  # 根据实际数据结构调整reset_index逻辑
    
    # 定义PyArrow Schema，强制指定兼容Spark的类型
    # valid_time: 映射为timestamp(ns)，对应Spark的timestamp类型
    # latitude: 映射为float64，对应Spark的double类型（显式指定避免元数据推断偏差）
    schema = pa.schema([
        ('valid_time', pa.timestamp('ns')),
        ('latitude', pa.float64()),
        # 其他列按实际需求添加，示例：
        # ('longitude', pa.float64()),
        # ('temperature', pa.float32())
    ])
    
    # 转换DataFrame为PyArrow Table并应用自定义schema
    table = pa.Table.from_pandas(df, schema=schema)
    
    # 写入Parquet，可选snappy压缩平衡性能与存储
    pq.write_table(table, parquet_output_path, compression='snappy')

3. 关键说明 #

• datetime类型兼容：PyArrow的pa.timestamp('ns')会将Pandas的datetime64[ns]序列化为Parquet标准的timestamp类型，Spark 3.5.0读取时会自动识别为timestamp，彻底解决类型报错问题。
• float类型兼容：Pandas的float64本身等价于Spark的double，显式指定schema是为了避免Parquet元数据自动推断时出现意外偏差。
• 性能保障：全程用Pandas+PyArrow处理，比Spark DataFrame转换更轻量，PyArrow的写入效率也能匹配主程序依赖RDD的性能要求。

替代方案（无PyArrow依赖时） #

如果无法使用PyArrow，可将datetime列转换为纳秒级整数，写入后在Spark读取时再转回timestamp：

def process_my_file(netcdf_path, parquet_output_path):
    ds = xr.open_dataset(netcdf_path)
    df = ds.to_dataframe().reset_index()
    
    # 将datetime转为纳秒整数存储
    df['valid_time'] = df['valid_time'].astype('int64')
    
    # 用默认引擎写入Parquet
    df.to_parquet(parquet_output_path, compression='snappy')

# Spark读取时转换类型（Scala示例）
val spark_df = spark.read.parquet("/path/to/parquet")
val final_df = spark_df.withColumn("valid_time", $"valid_time".cast("timestamp"))

这种方案无需额外依赖，但需要在Spark侧补充类型转换，适合对依赖严格限制的场景。

验证方法 #

写入后用Spark读取验证schema：

val df = spark.read.parquet("/path/to/parquet")
df.printSchema()
// 预期输出：
// root
//  |-- valid_time: timestamp (nullable = true)
//  |-- latitude: double (nullable = true)

内容的提问来源于stack exchange，提问作者Kelly Ma