PySpark SQL/Hive创建Parquet表文件格式及版本异常疑问

问题背景 #

在Hadoop集群（Spark 2.4.5、Hive 3.1.2）中遇到以下现象：

• 使用PySpark SQL创建并插入Parquet表：

  spark.sql("CREATE TABLE my_table (...) PARTITIONED BY (...) STORED AS Parquet")
  spark.sql("INSERT INTO my_table SELECT * FROM my_other_table")
  

  生成的文件缺少.snappy.parquet扩展名；
• Hive中执行相同的建表+插入操作，同样存在无扩展名的问题；
• 但使用PySpark DataFrame的saveAsTable方法：

  df.write.partitionBy("my_column").saveAsTable(name="my_table", format="Parquet")
  

  创建的表文件正常带有.snappy.parquet扩展名。

补充信息（2022年12月27日） #

集群无parquet-tools，通过HDFS命令检查发现所有文件首尾均有PAR1标识，各操作的Parquet元数据统计如下：

Method                      # of files      Total size  Parquet version                 File name

Hive Insert                 8               34.7 G      Jparquet-mr version 1.10.0      xxxxxx_x
PySpark SQL Insert          8               10.4 G      Iparquet-mr version 1.6.0       part-xxxxx-xxxxxxxx-xxxx-xxxx-xxxx-xxxxxxxxxxxx.c000
PySpark DF insertInto       8               10.9 G      Iparquet-mr version 1.6.0       part-xxxxx-xxxxxxxx-xxxx-xxxx-xxxx-xxxxxxxxxxxx.c000
PySpark DF saveAsTable      8               11.5 G      Jparquet-mr version 1.10.1      part-xxxxx-xxxxxxxx-xxxx-xxxx-xxxx-xxxxxxxxxxxx-c000.snappy.parquet

注：为保证生成文件数量一致，DataFrame操作使用repartition，SQL操作使用distribute by

疑问解答 #

1. 为何4种情况中有3种无文件扩展名？ #

这是不同写入路径的实现逻辑差异导致的：

• saveAsTable属于Spark DataFrame API的原生写入路径，Spark对该路径的Parquet写入做了标准化处理，默认会添加.snappy.parquet扩展名；
• PySpark SQL的INSERT INTO、Hive插入、DataFrame的insertInto均走Hive兼容写入路径，依赖Hive的parquet-mr写入器。该写入器在生成文件时不会强制添加规范的Parquet扩展名，仅生成简单文件名或分段标识（如.c000后缀）；
• 扩展名缺失不影响Parquet文件的正常读取，仅为文件名规范层面的区别。

2. 为何Hive插入生成的文件体积过大？ #

你的推测正确，核心原因是Hive未启用Snappy压缩：

• Hive默认未开启Parquet压缩，需手动配置以下参数启用Snappy：

  hive.exec.compress.output=true
  mapreduce.output.fileoutputformat.compress.codec=org.apache.hadoop.io.compress.SnappyCodec
  parquet.compression=SNAPPY
  

• 对比其他操作：PySpark的SQL、insertInto、saveAsTable默认均启用了Snappy压缩，因此文件体积远小于Hive的未压缩数据；
• Hive使用的parquet-mr 1.10.0版本在未配置压缩时，会直接写入未压缩的Parquet数据，导致体积大幅增加。

3. 为何PySpark SQL与DataFrame的Parquet版本/实现存在差异，如何显式设置？ #

差异源于Spark使用的Parquet写入引擎不同：

• PySpark SQL的INSERT INTO、DataFrame的insertInto使用的是Hive的parquet-mr引擎（Spark 2.4.x默认绑定的兼容版本为1.6.0）；
• DataFrame的saveAsTable默认使用Spark原生Parquet引擎（即parquet-hadoop，版本1.10.1，由Spark独立维护）；
• 可通过以下Spark配置参数统一引擎与版本：
  • 统一使用Spark原生Parquet引擎（推荐，兼容性与性能更优）：

    spark.sql.parquet.writeLegacyFormat=false
    spark.sql.hive.convertMetastoreParquet=true
    

    注：spark.sql.hive.convertMetastoreParquet默认值为true，若被修改为false会强制走Hive引擎
  • 统一使用Hive的parquet-mr引擎：

    spark.sql.parquet.writeLegacyFormat=true
    spark.sql.hive.convertMetastoreParquet=false
    

    同时可通过parquet.version=1.10参数指定Parquet版本，对齐Hive使用的版本。

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