使用PySpark Parquet迁移SQL Server至BigQuery的datetime类型适配问题

你在通过 PySpark 将 SQL Server 数据导出为 GCS 上的 Parquet 文件，再加载到 BigQuery 时遇到了类型不匹配错误：目标 BigQuery 表的 DateAdded 列是 DATETIME 类型，但 Parquet 文件中对应列是 TIMESTAMP 类型，导致加载失败。以下是无需手动逐列修改的两种解决方案：

一、PySpark 抽取阶段（从源头对齐类型，推荐） #

SQL Server 的 datetime 类型经 JDBC 读取后会被 PySpark 转为 TimestampType，默认写入 Parquet 时仍为 TIMESTAMP 类型；而 BigQuery 的 DATETIME 是无时区的日期时间类型，TIMESTAMP 则带时区属性。可以通过配置或自动转换让输出的 Parquet 列匹配 BigQuery 要求：

方法1：全局配置 Spark 对齐时区与输出格式 #

在初始化 SparkSession 时添加以下配置，确保时间值和类型兼容：

from pyspark.sql import SparkSession
from pyspark.sql.types import *

spark = SparkSession.builder \
    .appName("SQLServerToGCS") \
    .config("spark.jars", r"path\to\mssql-jdbc-12.8.1.jre8.jar, path\to\gcs-connector-hadoop3-latest.jar") \
    .config("spark.hadoop.google.cloud.auth.service.account.enable", "true") \
    .config("spark.hadoop.google.cloud.auth.service.account.json.keyfile", r"path\to\sa.json") \
    .config("spark.hadoop.fs.gs.impl", "com.google.cloud.hadoop.fs.gcs.GoogleHadoopFileSystem") \
    .config("spark.sql.parquet.int96RebaseModeInWrite", "CORRECTED") \
    # 新增：设置为 SQL Server 数据库使用的时区（示例为北京时间）
    .config("spark.sql.session.timeZone", "Asia/Shanghai") \
    # 新增：将 Timestamp 转为 BigQuery 可识别为 DATETIME 的字符串格式
    .config("spark.sql.parquet.outputTimestampType", "STRING") \
    .config("spark.driver.memory", "32g") \
    .getOrCreate()

说明：

• spark.sql.session.timeZone 必须与 SQL Server 时区一致，避免时间偏移；
• spark.sql.parquet.outputTimestampType 设置为 STRING 后，PySpark 会将 Timestamp 以 yyyy-MM-dd HH:mm:ss 格式写入 Parquet，BigQuery 加载时会自动映射为 DATETIME 类型。

方法2：批量自动转换 Timestamp 列 #

如果不想修改全局配置，可在写入 Parquet 前批量转换所有 Timestamp 类型列：

from pyspark.sql.functions import date_format

# 自动识别所有 Timestamp 类型的列
timestamp_cols = [col.name for col in df.schema if isinstance(col.dataType, TimestampType)]

# 批量转为 yyyy-MM-dd HH:mm:ss 格式的字符串
for col_name in timestamp_cols:
    df = df.withColumn(col_name, date_format(col_name, "yyyy-MM-dd HH:mm:ss"))

# 写入 Parquet
df.write.mode("overwrite").parquet(gcs_bucket)

二、BigQuery 加载阶段（无需修改已有 Parquet 文件） #

如果已经生成了 Parquet 文件，可在 LoadJob 中配置 schema 映射，强制将 Parquet 的 TIMESTAMP 类型转换为目标表的 DATETIME 类型：

from google.cloud import bigquery
from google.cloud import storage

# 原有配置不变
bucket_name = "bucket_name"
file_prefix = "file_prefix"
project_id = "bq_project_id"
dataset_id = "dataset"
table_id = "table"

bq_client = bigquery.Client()
storage_client = storage.Client()
bucket = storage_client.bucket(bucket_name)
blobs = bucket.list_blobs(prefix=file_prefix)
table_ref = f"{project_id}.{dataset_id}.{table_id}"
files = sorted([blob.name for blob in blobs if blob.name.endswith('.parquet')])

for file_name in files:
    file_path = f"gs://bucket_name/{file_name}"
    print(f"Processing file: {file_path}")
    
    # 获取目标表的 schema 并修改类型映射
    target_table = bq_client.get_table(table_ref)
    adjusted_schema = [
        bigquery.SchemaField(
            field.name,
            # 将目标表中 DATETIME 类型的列强制声明，让 BigQuery 自动转换源 TIMESTAMP
            "DATETIME" if field.field_type == "DATETIME" else field.field_type,
            mode=field.mode
        ) for field in target_table.schema
    ]
    
    # 配置 LoadJob
    job_config = bigquery.LoadJobConfig(
        source_format=bigquery.SourceFormat.PARQUET,
        write_disposition=bigquery.WriteDisposition.WRITE_APPEND,
        ignore_unknown_values=True,
        schema=adjusted_schema,
        autodetect=False  # 关闭自动检测，使用指定 schema
    )
    
    load_job = bq_client.load_table_from_uri(file_path, table_ref, job_config=job_config)
    load_job.result()
    print(f"Loaded file {file_name} into BigQuery table {table_ref}.")

说明：

• 需确保 BigQuery 数据集的时区与 SQL Server 一致，避免时间值偏移；
• 通过 schema 参数强制指定目标列类型，BigQuery 会自动完成 TIMESTAMP 到 DATETIME 的转换。

验证提示：
无论采用哪种方案，建议先导出少量数据测试，检查日期时间值是否准确无偏移。

内容的提问来源于stack exchange，提问作者Dina Kleper