Scala-Spark 扁平化嵌套模式包含数组

Question

我有一个包含数组的嵌套模式：

 root
  |-- alarm_time: string (nullable = true)
  |-- alarm_id: string (nullable = true)
  |-- user: struct (nullable = true)
  |    |-- name: string (nullable = true)
  |    |-- family: string (nullable = true)
  |    |-- address: struct (nullable = true)
  |    |    |-- postalcode: string (nullable = true)
  |    |    |-- line1: string (nullable = true)
  |    |    |-- city: string (nullable = true)
  |    |    |-- country: string (nullable = true) 
  |-- device: struct (nullable = true)
  |    |-- device_usage: string (nullable = true)
  |    |-- device_id: string (nullable = true)  
  |-- alarm_info: struct (nullable = true)
  |    |-- type: string (nullable = true)
  |    |-- reason: string (nullable = true)
  |    |-- data: struct (nullable = true)
  |    |    |-- alarm_severity: long (nullable = true)
  |    |    |-- extra_info: array (nullable = true)
  |    |    |    |-- element: struct (containsNull = true)
  |    |    |    |    |-- producer: string (nullable = true)
  |    |    |    |    |-- comment: string (nullable = true)

我过去常常忽略数组字段并使用此代码来扁平化我的架构：

def flattenSchema(schema: StructType, prefix: String = null) : Array[Column] = {
  schema.fields.flatMap(f => {
    val colName = if (prefix == null) f.name else (prefix + "." + f.name)

    f.dataType match {
      case st: StructType => flattenSchema(st, colName)
      case _ => Array(col(colName))
    }
  })
}

并像df.select(flattenSchema(df.schema):_*) 一样使用它，但现在我有一个用例也需要保留数组数据，我唯一能想到的就是分解数组并保留多行但我没有运气。由于我将列作为 args 参数传递，因此我无法传递另一个参数。

我怎样才能做到这一点（使用展开数组的扁平化架构）？

Answer 1

所以我现在（Spark 2.2+）正在做的是检查架构是否嵌套并一遍又一遍地调用flattenschema，直到它变平。

  def makeItFlat(df: DataFrame): DataFrame = {
    if (isSchemaNested(df)) {
      val flattenedSchema = flattenSchema(df.schema)
      makeItFlat(df.select(flattenedSchema: _*))
    }
    else {
      df
    }
  }

makeItFlat() 是一种递归方法，用于检查模式是否未展平但再次递归调用 flattenschema

  def isSchemaNested(df: DataFrame): Boolean = {
    df.schema.fields.flatMap(field => {

      field.dataType match {
        case arrayType: ArrayType => {
          Array(true)
        }
        case mapType: MapType => {
          Array(true)
        }
        case structType: StructType => {
          Array(true)
        }
        case _ => {
          Array(false)
        }
      }
    }).exists(b => b)
  }

isSchemaNested 的工作是检查模式定义中是否存在任何嵌套数据类型

  private def flattenSchema(schema: StructType, prefix: String = null): Array[Column] = {
    schema.fields.flatMap(field => {
      val columnName = if (prefix == null) field.name else prefix + "." + field.name

      field.dataType match {
        case arrayType: ArrayType => {

          Array[Column](explode_outer(col(columnName)).as(columnName.replace(".", "_")))
        }
        case mapType: MapType => {
          None
        }
        case structType: StructType => {

          flattenSchema(structType, columnName)
        }
        case _ => {
          val columnNameWithUnderscores = columnName.replace(".", "_")

          val metadata = new MetadataBuilder().putString("encoding", "ZSTD").build()

          Array(col(columnName).as(columnNameWithUnderscores, metadata))
        }
      }
    }).filter(field => field != None)
  }

Answer 2

Am1rr3zA，如果我们在同一级别有两个数组，您提供的解决方案将会中断。它不允许同时发生两个爆炸：“每个选择子句只允许一个生成器，但发现了 2 个：explode(_1),explode(_2)”

我已更新解决方案以跟踪嵌套中的复杂类型

  def flattenDataFrame(df: DataFrame): DataFrame = {

var flattenedDf: DataFrame = df

if (isNested(df)) {

  val flattenedSchema: Array[(Column, Boolean)] = flattenSchema(df.schema)
  var simpleColumns: List[Column] = List.empty[Column]
  var complexColumns: List[Column] = List.empty[Column]

  flattenedSchema.foreach {
    case (col, isComplex) => {
      if (isComplex) {
        complexColumns = complexColumns :+ col
      } else {
        simpleColumns = simpleColumns :+ col
      }
    }
  }

  var crossJoinedDataFrame = df.select(simpleColumns: _*)
  complexColumns.foreach(col => {
    crossJoinedDataFrame = crossJoinedDataFrame.crossJoin(df.select(col))
    crossJoinedDataFrame = flattenDataFrame(crossJoinedDataFrame)
  })
  crossJoinedDataFrame
} else {
  flattenedDf
}

}

private def flattenSchema(schema: StructType, prefix: String = null): Array[(Column, Boolean)] = {

schema.fields.flatMap(field => {

  val columnName = if (prefix == null) field.name else prefix + "." + field.name
  field.dataType match {
    case arrayType: ArrayType => {
      val cols: Array[(Column, Boolean)] = Array[(Column, Boolean)](((explode_outer(col(columnName)).as(columnName.replace(".", "_"))), true))
      cols
    }
    case structType: StructType => {
      flattenSchema(structType, columnName)
    }
    case _ => {
      val columnNameWithUnderscores = columnName.replace(".", "_")
      val metadata = new MetadataBuilder().putString("encoding", "ZSTD").build()
      Array(((col(columnName).as(columnNameWithUnderscores, metadata)), false))
    }
  }
}).filter(field => field != None)

}

def isNested(df: DataFrame): Boolean = {
df.schema.fields.flatMap(field => {

  field.dataType match {
    case arrayType: ArrayType => {
      Array(true)
    }
    case mapType: MapType => {
      Array(true)
    }
    case structType: StructType => {
      Array(true)
    }
    case _ => {
      Array(false)
    }
  }
}).exists(b => b)

}