json - 合并不区分大小写的 json 列名称

Question

我的 JSON 列名称是大小写的组合(例如:title/Title 和 name/Name)，因此在输出中，我得到name 和 Name 作为两个不同的列(类似于 title 和 Title)。

如何使 JSON 列不区分大小写？

config("spark.sql.caseSensitive", "true") -> 我试过了，但没用。

val df = Seq(
  ("A", "B", "{\"Name\":\"xyz\",\"Address\":\"NYC\",\"title\":\"engg\"}"),
  ("C", "D", "{\"Name\":\"mnp\",\"Address\":\"MIC\",\"title\":\"data\"}"),
  ("E", "F", "{\"name\":\"pqr\",\"Address\":\"MNN\",\"Title\":\"bi\"}")
)).toDF("col_1", "col_2", "col_json")

import sc.implicits._
val col_schema = spark.read.json(df.select("col_json").as[String]).schema

val outputDF = df.withColumn("new_col", from_json(col("col_json"), col_schema))
       .select("col_1", "col_2", "new_col.*")

outputDF.show(false)

当前输出:

预期/需要的输出(列名不区分大小写):

Answer 1

解决方案 1

您可以按小写名称对列进行分组，并使用 coalesce 函数合并它们:

// set spark.sql.caseSensitive to true to avoid ambuigity
spark.conf.set("spark.sql.caseSensitive", "true")

val col_schema = spark.read.json(df.select("col_json").as[String]).schema

val df1 = df.withColumn("new_col", from_json(col("col_json"), col_schema))
     .select("col_1", "col_2", "new_col.*")


val mergedCols = df1.columns.groupBy(_.toLowerCase).values
  .map(grp =>
    if (grp.size > 1) coalesce(grp.map(col): _*).as(grp(0))
    else col(grp(0))
  ).toSeq

val outputDF = df1.select(mergedCols:_*)

outputDF.show()
//+----+-------+-----+-----+-----+
//|Name|Address|col_1|Title|col_2|
//+----+-------+-----+-----+-----+
//|xyz |NYC    |A    |engg |B    |
//|mnp |MIC    |C    |data |D    |
//|pqr |MNN    |E    |bi   |F    |
//+----+-------+-----+-----+-----+

解决方案2

另一种方法是将 JSON 字符串列解析为 MapType 而不是 StructType，并使用 transform_keys您可以将列名小写，然后展开 map 并旋转以获取列:

import org.apache.spark.sql.types.{MapType, StringType}

val outputDF = df.withColumn(
    "col_json",
    from_json(col("col_json"), MapType(StringType, StringType))
  ).select(
    col("col_1"),
    col("col_2"),
    explode(expr("transform_keys(col_json, (k, v) -> lower(k))"))
  ).groupBy("col_1", "col_2")
  .pivot("key")
  .agg(first("value"))

outputDF.show()
//+-----+-----+-------+----+-----+
//|col_1|col_2|address|name|title|
//+-----+-----+-------+----+-----+
//|E    |F    |MNN    |pqr |bi   |
//|C    |D    |MIC    |mnp |data |
//|A    |B    |NYC    |xyz |engg |
//+-----+-----+-------+----+-----+

对于此解决方案，transform_keys 仅适用于 Spark 3，对于旧版本，您可以使用 UDF:

val mapKeysToLower = udf((m: Map[String, String]) => {
  m.map { case (k, v) => k.toLowerCase -> v }
})