PySpark执行Compaction后Hive分区数据异常问题求助

可能的原因 #

• 直接写入Hive ACID事务表的底层目录：如果你的Hive表是ACID表（开启了transactional=true），Spark直接写底层目录会破坏ACID的事务日志和数据文件结构，Hive读取时解析错误，导致数据错乱。
• 读写格式不匹配：Hive表的存储格式（如ORC、Parquet）和Spark读写时指定的格式不一致。比如Hive表用ORC存储，但Spark读取/写入时用了默认的CSV，导致数据解析/写入时列错位、类型不匹配。
• 未清理旧文件就写入：写回原目录时没有先删除旧数据文件，新旧文件共存且格式不兼容，Hive读取时混合解析导致数据异常。
• Schema推断错误：Spark读取数据时自动推断Schema，和Hive表的实际Schema（列顺序、数据类型）不匹配，导致写入后数据列错位。比如原表列顺序是Column A、Column B，但Spark推断时颠倒了顺序，写入后数据位置偏移。

修复方案 #

针对ACID表的情况 #

• 不要直接操作Hive ACID表的底层目录，改用Spark通过Hive Catalog操作表：

  # 读取指定分区数据
  df = spark.table("database.table_name").filter("partition_col='your_partition_value'")
  # 重分区后写入临时表
  df.coalesce(target_file_num).write.mode("overwrite").saveAsTable("database.temp_table")
  # 通过Hive INSERT OVERWRITE覆盖原分区
  spark.sql("INSERT OVERWRITE TABLE database.table_name PARTITION(partition_col='your_partition_value') SELECT * FROM database.temp_table")
  

• 若无需ACID特性，可修改表属性关闭事务：ALTER TABLE database.table_name SET TBLPROPERTIES ('transactional'='false')，再执行压缩操作。

针对读写格式不匹配的情况 #

• 明确指定与Hive表一致的存储格式和压缩编码：

  # 以ORC格式+snappy压缩为例
  df = spark.read.format("orc").load("hdfs://path/to/partition")
  df.coalesce(target_file_num).write.format("orc").option("compression", "snappy").mode("overwrite").save("hdfs://path/to/partition")
  

• 提前确认Hive表存储格式：执行DESCRIBE FORMATTED database.table_name，查看Storage Desc Params中的inputFormat和outputFormat。

清理旧文件后写入 #

• 写入前递归删除原目录下的旧数据文件（保留分区目录）：

  hadoop_conf = spark.sparkContext._jsc.hadoopConfiguration()
  fs = spark.sparkContext._jvm.org.apache.hadoop.fs.FileSystem.get(hadoop_conf)
  path = spark.sparkContext._jvm.org.apache.hadoop.fs.Path("hdfs://path/to/partition")
  if fs.exists(path):
      for file_status in fs.listStatus(path):
          if not file_status.isDirectory():
              fs.delete(file_status.getPath(), True)
  # 执行写入
  df.coalesce(target_file_num).write.format("orc").option("compression", "snappy").mode("append").save("hdfs://path/to/partition")
  

• 也可直接使用Spark的mode("overwrite")，但需注意该操作会删除整个分区目录，需确保分区参数正确。

避免Schema推断错误 #

• 读取时手动指定与Hive表一致的Schema，不依赖自动推断：

  from pyspark.sql.types import StructType, StructField, IntegerType
  schema = StructType([
      StructField("Column A", IntegerType(), nullable=True),
      StructField("Column B", IntegerType(), nullable=True)
  ])
  df = spark.read.schema(schema).format("orc").load("hdfs://path/to/partition")
  

• 或通过Hive Catalog直接读取表，自动获取正确Schema：df = spark.table("database.table_name").filter("partition_col='your_partition_value'")

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