Spark Dataframe验证parquet写入的列名

Question

我正在使用从JSON事件流转换而来的Dataframes处理事件，这些数据流最终被写为Parquet格式。

但是，有些JSON事件在键中包含空格，我希望在将数据帧转换为Parquet之前记录并过滤/删除这些事件，因为；{}()\n\t=被认为是Parquet模式（CatalystSchemaConverter）中的特殊字符，如下面[1]所列，因此列名中不应允许。

我如何在Dataframe中对列名进行这样的验证，并在不出错的情况下完全删除这样的事件。

[1]Spark's 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)