提问者:小点点

Spark中区分大小写的拼花架构合并


我正在尝试使用Spark加载和分析一些镶木地板文件。我正在使用schemaMerge来加载文件,因为较新的文件有一些额外的列。此外,一些文件的列名为小写,其他文件为大写。

举个例子

file1. parque有一个类似的模式

column1 integer,
column2 integer

file2. parque类似:

Column1 integer,
Column2 integer,
Column3 integer

我遇到了ParquetFileFormat类的INF Schema方法的问题。Schema合并被委托给Spark sql的StructTypemerge方法。据我所知,该方法只能以区分大小写的方式工作。在内部,它使用映射按名称查找字段,如果情况不匹配,它会将其解释为新字段。稍后,当检查模式是否重复时,尊重区分大小写的配置,我们最终会拥有重复的列。这导致

org.apache.spark.sql.AnalysisException: Found duplicate column(s) in the data schema

有没有办法让架构合并不区分大小写?

我希望得到这样的结果模式:

column1 integer,
column2 integer,
Column3 integer

共1个答案

匿名用户

您可以在配置中设置lark. sql.caseSen的=true使SparkSQL模式区分大小写。它也会影响模式合并。

scala> spark.conf.set("spark.sql.caseSensitive","true")

scala> val df = sc.parallelize(1 to 1000).toDF()
df: org.apache.spark.sql.DataFrame = [value: int]

scala> df.withColumnRenamed("value","VALUE").write.parquet("test_uc")

scala> df.write.parquet("test_lc")

scala> val df2=spark.read.option("mergeSchema","true").parquet("test_*")
df2: org.apache.spark.sql.DataFrame = [value: int, VALUE: int]

scala> val merged = df2.columns.groupBy(_.toLowerCase)
                   .map(t => coalesce(t._2.map(col):_*).as(t._1))
                   .toArray
merged: Array[org.apache.spark.sql.Column] = Array(coalesce(value, VALUE) AS `value`)

scala> df2.select(merged:_*)
res2: org.apache.spark.sql.DataFrame = [value: int]

scala> spark.conf.set("spark.sql.caseSensitive","false")

// process your dataframe