从PySpark向BigQuery TIME类型列写入数据失败求助

问题现象 #

向已有TIME类型列的BigQuery表写入Spark DataFrame时，报错Provided Schema does not match Table ... Field wake_up_time has changed type from TIME to STRING，即使字符串格式符合BigQuery TIME字面量要求。

解决方案 #

方案1：显式指定BigQuery Schema #

写入时通过schema参数强制映射DataFrame字段到BigQuery的TIME类型，覆盖连接器的自动Schema推断：

(
    df
    .write
    .format("bigquery")
    .option("temporaryGcsBucket", "temp_bucket")
    .option("table", "test_project.test_dataset.time_test")
    .option("writeMethod", "direct")
    # 明确指定字段对应的BigQuery类型
    .option("schema", "name:STRING,wake_up_time:TIME")
    .mode("append")
    .save()
)

方案2：将字符串转换为BigQuery TIME对应的Spark类型 #

BigQuery TIME类型在Spark中对应从午夜开始的微秒数（LongType），将时间字符串转换为该格式后写入：

from pyspark.sql.types import LongType
from pyspark.sql.functions import col

# 定义转换函数：将HH:MM:SS转为微秒数
def time_str_to_micros(time_str):
    hours, mins, secs = map(int, time_str.split(":"))
    return (hours * 3600 + mins * 60 + secs) * 1_000_000

# 注册UDF并转换字段
time_to_micros_udf = spark.udf.register("time_to_micros", time_str_to_micros, LongType())
df_converted = df.withColumn("wake_up_time", time_to_micros_udf(col("wake_up_time")))

# 执行写入
(
    df_converted
    .write
    .format("bigquery")
    .option("temporaryGcsBucket", "temp_bucket")
    .option("table", "test_project.test_dataset.time_test")
    .option("writeMethod", "direct")
    .mode("append")
    .save()
)

原因解析 #

官方文档中"字符串符合BQ TIME格式可写入"的描述，适用于自动创建BigQuery表的场景：连接器会将符合格式的字符串字段推断为TIME类型。但当目标表已存在时，连接器会严格匹配Spark DataFrame的Schema类型与BQ表类型，此时Spark的StringType会被映射为BQ的STRING类型，与已有表的TIME类型冲突，触发Schema不匹配错误。

通过显式指定Schema或转换为对应微秒数的LongType，可以让连接器正确识别字段类型，完成写入操作。

内容的提问来源于stack exchange，提问作者Waqar Ahmed