PySpark Oracle导HDFS Parquet遇年份越界错误的解决咨询

核心原因 #

Parquet格式对应的Spark DateType有严格范围限制：仅支持1582-10-15（格里高利历起始）至9999-12-31的日期值，0001-01-01完全超出这个范围，因此写入时触发异常。

可行解决方案 #

1. 将日期字段转为字符串类型写入 #

如果业务允许日期以字符串形式存储，这是最直接的方案，彻底规避日期范围限制：

from pyspark.sql.functions import col

# 假设原DataFrame为df，日期列名为`create_time`
df_processed = df.withColumn("create_time", col("create_time").cast("string"))
# 写入Parquet
df_processed.write.mode("overwrite").parquet("hdfs://path/to/target")

2. 替换非法日期为合法值 #

如果必须保留日期类型，可将超出范围的日期替换为NULL或业务允许的默认日期（比如1582-10-15）：

from pyspark.sql.functions import when, lit

# 检查日期是否在合法范围内，非法则替换为NULL
df_processed = df.withColumn(
    "create_time",
    when(
        col("create_time").between(lit("1582-10-15"), lit("9999-12-31")),
        col("create_time")
    ).otherwise(lit(None))
)
# 写入Parquet
df_processed.write.mode("overwrite").parquet("hdfs://path/to/target")

3. 在ThreadPool任务中添加局部异常处理 #

由于使用ThreadPool并行处理表导入，需确保单个任务的异常不会终止整个线程池，可在每个导入任务内部捕获异常并记录：

from concurrent.futures import ThreadPoolExecutor
import logging

def import_table(table_name):
    try:
        # 读取Oracle表数据
        df = spark.read.format("jdbc") \
            .option("url", "jdbc:oracle:thin:@//host:port/service") \
            .option("dbtable", table_name) \
            .option("user", "user") \
            .option("password", "pass") \
            .load()
        
        # 加入日期处理逻辑（转字符串或替换非法值）
        df_processed = df.withColumn("create_time", col("create_time").cast("string"))
        
        # 写入Parquet
        df_processed.write.mode("overwrite").parquet(f"hdfs://path/to/target/{table_name}")
        logging.info(f"成功导入表 {table_name}")
    except Exception as e:
        logging.error(f"导入表 {table_name} 失败: {str(e)}")
        # 可选：记录失败表名，后续重试
        with open("failed_tables.txt", "a") as f:
            f.write(f"{table_name}\n")

# 并行执行
with ThreadPoolExecutor(max_workers=5) as executor:
    executor.map(import_table, ["table1", "table2", "table3"])

注意事项 #

• 不要尝试修改Spark或Parquet的底层日期范围限制，这是格式本身的规范，强行修改会导致数据损坏或兼容性问题。
• 处理日期前建议先统计非法日期的数量和分布，确认业务是否接受替换/转字符串的方案。

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