Spark写入分区Hive表时分区规格错误的正确解决方法

问题背景 #

我有一张分区Hive表，定义如下：

CREATE TABLE IF NOT EXISTS TRADES
(
    TRADE_ID                          STRING,
    TRADE_DATE                        INT,
-- ...
)
    PARTITIONED BY (BUSINESS_DATE INT)
    STORED AS PARQUET;

通过Spark Java应用插入数据的代码如下：

try (SparkSession sparkSession = SparkSession.builder()
        .config(new SparkConf())
        .enableHiveSupport()
//        .config("hive.exec.dynamic.partition", "true")
//        .config("hive.exec.dynamic.partition.mode", "nonstrict")
//        .config("spark.sql.hive.convertMetastoreParquet", "false")
        .getOrCreate()) {

  dataset.select(columns(joinedDs, businessDate))
        .write()
        .format("parquet")
        .option("compression", "snappy")
        .mode(SaveMode.Append)
        .insertInto("TRADES"));
}

//...

private Column[] columns(Dataset<Row> dataset, LocalDate businessDate) {
  List<Column> columns = new ArrayList<>();
  for (String column : appConfig.getColumns()) {
    columns.add(dataset.col(column));
  }
  columns.add(lit(dateToInteger(businessDate)).as("BUSINESS_DATE"));
  return columns.toArray(new Column[0]);
}

执行时触发异常：

23/01/27 10:39:26 ERROR metadata.Hive: Exception when loading partition with parameters  partPath=hdfs://path-to-trades/.hive-staging_hive_2023-01-27_10-38-29_374_384898966661095068-1/-ext-10000/BUSINESS_DATE=20221230,  table=trades,  partSpec={business_date=, BUSINESS_DATE=20221230},  replace=false,  listBucketingEnabled=false,  isAcid=false,  hasFollowingStatsTask=false
org.apache.hadoop.hive.ql.metadata.HiveException: MetaException(message:Partition spec is incorrect. {business_date=, BUSINESS_DATE=20221230})
    at org.apache.hadoop.hive.ql.metadata.Hive.loadPartitionInternal(Hive.java:1662)
    at org.apache.hadoop.hive.ql.metadata.Hive.lambda$loadDynamicPartitions$4(Hive.java:1970)
    at java.util.concurrent.FutureTask.run(FutureTask.java:266)
    at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
    at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
    at java.lang.Thread.run(Thread.java:750)
Caused by: MetaException(message:Partition spec is incorrect. {business_date=, BUSINESS_DATE=20221230})
    at org.apache.hadoop.hive.metastore.Warehouse.makePartName(Warehouse.java:329)
    at org.apache.hadoop.hive.metastore.Warehouse.makePartPath(Warehouse.java:312)
    at org.apache.hadoop.hive.ql.metadata.Hive.genPartPathFromTable(Hive.java:1751)
    at org.apache.hadoop.hive.ql.metadata.Hive.loadPartitionInternal(Hive.java:1607)
    ... 5 more

我通过将分区列统一改为小写的方式临时解决了问题：
修改后的表DDL：

CREATE TABLE IF NOT EXISTS TRADES
(
    TRADE_ID                          STRING,
    TRADE_DATE                        INT,
-- ...
)
    PARTITIONED BY (business_date INT) -- <-- 小写！
    STORED AS PARQUET;

修改后的Java代码：

columns.add(lit(dateToInteger(businessDate)).as("business_date")); // <-- 小写！

我推测可以通过配置Spark或Hive实现大小写兼容，无需修改代码和表结构，但尝试启用注释中的Spark配置项后并未解决问题。请问在Spark 2.4.0和Hive 2.1.1环境下，如何正确实现该需求？

P.S. sparkSession.sql("describe formatted TRADES").show(false)的输出如下：

+--------------------+--------------------+-------+
|            col_name|           data_type|comment|
+--------------------+--------------------+-------+
|           ROW_INDEX|                 int|   null|
|  OP_GENESIS_FEED_ID|              string|   null|
|       business_date|                 int|   null|
|# Partition Infor...|                    |       |
|          # col_name|           data_type|comment|
|       business_date|                 int|   null|
|                    |                    |       |
|# Detailed Table ...|                    |       |
|            Database|             managed|       |
|               Table|              trades|       |
|               Owner|             managed|       |
|        Created Time|Mon Jan 30 19:59:...|       |
|         Last Access|Thu Jan 01 02:00:...|       |
|          Created By|Spark 2.4.0-cdh6.2.1|       |
|                Type|             MANAGED|       |
|            Provider|                hive|       |
|    Table Properties|[transient_lastDd...|       |
|            Location|....................|       |
|       Serde Library|org.apache.hadoop...|       |
|         InputFormat|org.apache.hadoop...|       |
|        OutputFormat|org.apache.hadoop...|       |
|  Storage Properties|[serialization.fo...|       |
|  Partition Provider|             Catalog|       |
+--------------------+--------------------+-------+

解决方案 #

问题根源在于Hive元数据默认对列名大小写不敏感，但Spark处理分区列的大小写逻辑与Hive不匹配，导致元数据中出现重复的分区列记录（小写business_date和大写BUSINESS_DATE）。以下是无需修改表结构和代码的配置方案：

1. 配置Hive元数据大小写不敏感 #

在Hive的hive-site.xml中添加以下配置：

<property>
  <name>hive.metastore.table.name.case.sensitive</name>
  <value>false</value>
</property>
<property>
  <name>hive.metastore.column.name.case.sensitive</name>
  <value>false</value>
</property>
<property>
  <name>hive.metastore.schema.verification</name>
  <value>false</value>
</property>

这些配置确保Hive元数据对表名、列名的大小写不敏感，避免生成重复的分区列记录。

2. 配置Spark与Hive交互的大小写处理 #

在Spark应用的Session构建中添加以下配置：

SparkSession sparkSession = SparkSession.builder()
        .config(new SparkConf())
        .enableHiveSupport()
        .config("hive.exec.dynamic.partition", "true")
        .config("hive.exec.dynamic.partition.mode", "nonstrict")
        .config("spark.sql.caseSensitive", "false")
        .config("spark.sql.hive.caseSensitive", "false")
        .config("spark.sql.hive.convertMetastoreParquet", "false")
        .getOrCreate();

• spark.sql.caseSensitive：关闭Spark SQL自身的列名大小写敏感性
• spark.sql.hive.caseSensitive：确保Spark与Hive元数据交互时忽略大小写
• spark.sql.hive.convertMetastoreParquet：禁用Spark将Parquet表转换为内部格式，强制使用Hive的Parquet处理逻辑，消除大小写解析差异

3. 清理已有错误元数据（可选） #

如果Hive元数据中已经存在大小写不一致的分区列记录，需要手动清理：

ALTER TABLE TRADES DROP PARTITION (business_date='');

清理后重启Hive Metastore服务和Spark应用，再执行插入操作即可。

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