PySpark读写Parquet至BigQuery时timestamp类型转换错误解决方案咨询

错误原因 #

Parquet文件中timestamp字段实际以binary格式存储（可能是写入时采用了非标准编码），但Spark自动推断为timestamp类型，导致写入BigQuery时类型转换失败。

解决方案（保留BigQuery TIMESTAMP类型） #

1. 显式读取转换：先读为字符串再转标准Timestamp #

先定义读取Schema，将timestamp字段以StringType读取，再通过to_timestamp转换为Spark标准timestamp类型，确保写入BigQuery时类型匹配：

from pyspark.sql.types import StructType, StructField, StringType
from pyspark.sql.functions import to_timestamp

# 定义读取用Schema，将timestamp字段设为字符串类型
read_schema = StructType([
    StructField("event_timestamp_local", StringType(), nullable=True),
    StructField("event_timestamp_eu", StringType(), nullable=True),
    StructField("event_timestamp_us", StringType(), nullable=True),
    StructField("event_timestamp_asia", StringType(), nullable=True),
    StructField("event_timestamp_africa", StringType(), nullable=True)
])

# 读取ADLS上的Parquet文件
df = spark.read.schema(read_schema).parquet("abfss://<container>@<account>.dfs.core.windows.net/<path>")

# 转换为标准timestamp（根据实际时间格式调整format参数）
df_converted = df
for col_name in df.columns:
    df_converted = df_converted.withColumn(col_name, to_timestamp(df_converted[col_name], "yyyy-MM-dd HH:mm:ss"))

2. 调整Spark Parquet Timestamp解析配置 #

通过配置让Spark正确识别Parquet中以binary/int96存储的timestamp，避免自动推断错误：

# 开启int96格式timestamp转换（部分系统用int96存储timestamp）
spark.conf.set("spark.sql.parquet.int96TimestampConversion", "true")
# 禁用无时区timestamp，统一使用带时区的标准timestamp
spark.conf.set("spark.sql.parquet.timestampNTZ.enabled", "false")

# 再读取Parquet文件
df = spark.read.parquet("abfss://<container>@<account>.dfs.core.windows.net/<path>")

3. 写入BigQuery时显式指定Schema #

手动指定BigQuery目标表的Schema，确保和Spark DataFrame的timestamp类型完全匹配，避免自动推断出错：

import json

# 定义BigQuery目标Schema
bq_schema = [
    {"name": "event_timestamp_local", "type": "TIMESTAMP", "mode": "NULLABLE"},
    {"name": "event_timestamp_eu", "type": "TIMESTAMP", "mode": "NULLABLE"},
    {"name": "event_timestamp_us", "type": "TIMESTAMP", "mode": "NULLABLE"},
    {"name": "event_timestamp_asia", "type": "TIMESTAMP", "mode": "NULLABLE"},
    {"name": "event_timestamp_africa", "type": "TIMESTAMP", "mode": "NULLABLE"}
]

# 写入BigQuery
df_converted.write.format("bigquery") \
            .option("table", "<project-id>.<dataset-id>.<table-name>") \
            .option("schema", json.dumps(bq_schema)) \
            .mode("append")  # 按需选择mode：append/overwrite等
            .save()

内容的提问来源于stack exchange，提问作者Soumik Das