apache-spark - Spark Dataframe 验证 Parquet 写入的列名

Question

我正在使用从 JSON 事件流转换而来的数据帧处理事件，这些数据帧最终以 Parquet 格式写出。
但是，一些 JSON 事件在键中包含空格，我想在将其转换为 Parquet 之前从数据框中记录和过滤/删除此类事件，因为 ;{}()\n\t=在 中列出的 Parquet 模式 (CatalystSchemaConverter) 中被视为特殊字符[1] 下方 因此不应在列名中使用。
如何在 Dataframe 中对列名称进行此类验证并完全删除此类事件而不会出错 Spark Streaming 作业。
[1]
Spark 的 CatalystSchemaConverter

def checkFieldName(name: String): Unit = {
  // ,;{}()\n\t= and space are special characters in Parquet schema
  checkConversionRequirement(
    !name.matches(".*[ ,;{}()\n\t=].*"),
    s"""Attribute name "$name" contains invalid character(s) among " ,;{}()\\n\\t=".
             |Please use alias to rename it.
           """.stripMargin.split("\n").mkString(" ").trim
  )
}

Answer 1

对于在 中遇到此问题的每个人pyspark :在重命名列后，这甚至发生在我身上。经过一些迭代后，我可以让它工作的一种方法是:

file = "/opt/myfile.parquet"
df = spark.read.parquet(file)
for c in df.columns:
    df = df.withColumnRenamed(c, c.replace(" ", ""))

df = spark.read.schema(df.schema).parquet(file)