Spark Dataframe验证镶木地板写入的列名(标量) [英] Spark Dataframe validating column names for parquet writes (scala)
问题描述
我正在使用从JSON事件流转换而来的数据帧处理事件,这些数据最终以Parquet格式写出.
I'm processing events using Dataframes converted from a stream of JSON events which eventually gets written out as as Parquet format.
但是,某些JSON事件在键中包含空格,我想在将其转换为Parquet之前记录并过滤/删除数据框中的此类事件,因为,; {}()\ n \ t =被认为是特殊的如以下 [1] 中所列的Parquet模式(CatalystSchemaConverter)中的字符,因此不应在列名称中使用这些字符.
However, some of the JSON events contains spaces in the keys which I want to log and filter/drop such events from the data frame before converting it to Parquet because ,;{}()\n\t= are considered special characters in Parquet schema (CatalystSchemaConverter) as listed in [1] below and thus should not be allowed in the column names.
如何在Dataframe中对列名称进行此类验证,并完全删除此类事件,而又不会错失Spark Streaming作业.
How can I do such validations in Dataframe on the column names and drop such an event altogether without erroring out the Spark Streaming job.
[1] Spark的CatalystSchemaConverter
[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)
}
推荐答案
对于在 pyspark 中遇到此问题的每个人:重命名列后,这甚至都发生在我身上.在一些迭代之后,我可以使它工作的一种方法是:
For everyone experiencing this in pyspark: this even happened to me after renaming the columns. One way I could get this to work after some iterations is this:
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)
这篇关于Spark Dataframe验证镶木地板写入的列名(标量)的文章就介绍到这了,希望我们推荐的答案对大家有所帮助,也希望大家多多支持IT屋!