Databricks中PySpark读取Parquet文件报错：无法读取或转换文件架构

针对你遇到的java.io.IOException: Could not read or convert schema错误，提供以下排查和解决思路：

• 验证单个文件的合法性
  用PyArrow直接读取可疑Parquet文件，判断是文件损坏还是Spark兼容性问题。PyArrow的Parquet解析逻辑与Spark不同，能更精准定位问题：

  import pyarrow.parquet as pq
  
  try:
      # 替换为你的DBFS文件路径，注意转换为本地路径格式
      table = pq.read_table("/dbfs/.../xxx.parquet")
      print("文件Schema正常：")
      print(table.schema)
  except Exception as e:
      print(f"文件解析失败，确认损坏：{str(e)}")
  

  如果PyArrow也读失败，说明文件确实损坏，需要重新从Qlik Replicate导出对应文件；如果能正常读取，那就是Spark与Qlik导出的Parquet格式存在兼容性问题。

• 检查Qlik Replicate的导出配置
  Qlik Replicate导出Parquet时可能使用了Spark不兼容的设置：

  • 确认压缩格式是否为Spark支持的类型（如snappy、gzip，避免使用lz4的非标准变种）
  • 查看是否启用了特殊的Schema演化选项，或是否勾选了“Spark兼容”相关配置
  • 检查导出的Parquet版本，部分旧版本的元数据格式可能被Spark解析时拒绝

• 手动指定Schema读取
  跳过Spark自动Schema推断步骤，直接用源表的结构定义Schema，强制解析数据：

  from pyspark.sql.types import StructType, StructField, StringType, IntegerType, TimestampType
  
  # 根据你的源表结构自定义Schema
  custom_schema = StructType([
      StructField("user_id", IntegerType(), nullable=True),
      StructField("user_name", StringType(), nullable=True),
      StructField("update_time", TimestampType(), nullable=True)
  ])
  
  # 用指定Schema读取文件
  parquet_df = spark.read.format("parquet").schema(custom_schema).load("dbfs:/.../*.parquet")
  

  若能成功读取，说明问题出在文件的Schema元数据上，而非数据本身。

• 校验文件完整性
  对比ADLS容器中文件的大小与DBFS挂载后的文件大小，确认数据同步过程中没有出现截断或损坏：

  # 查看DBFS中文件的大小
  dbutils.fs.ls("dbfs:/.../xxx.parquet")
  

  如果大小不一致，重新挂载ADLS或检查Qlik Replicate的同步任务是否正常完成。

• 逐文件排查异常
  用binaryFile格式读取所有文件，再通过UDF逐个解析，定位具体损坏的文件和错误原因：

  from pyspark.sql.functions import udf
  from pyspark.sql.types import StringType
  import pyarrow.parquet as pq
  import io
  
  @udf(StringType())
  def parse_parquet(content):
      try:
          table = pq.read_table(io.BytesIO(content))
          return table.to_pandas().to_json(orient="records")
      except Exception as e:
          return f"解析错误：{str(e)}"
  
  # 读取所有Parquet文件为二进制
  binary_df = spark.read.format("binaryFile").load("dbfs:/.../*.parquet")
  # 解析并查看结果
  result_df = binary_df.withColumn("parsed_result", parse_parquet("content"))
  result_df.select("path", "parsed_result").show(truncate=False)
  

  从结果中筛选出有错误的文件，单独处理或重新导出。

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