如何遍历 Spark 中的模式？答案

【问题标题】：How to traverse through a schema in Spark?如何遍历 Spark 中的模式？
【发布时间】：2018-07-17 06:32:12
【问题描述】：

我想在 Spark 中迭代一个模式。使用df.schema 给出嵌套的StructType 和StructFields 的列表。

根元素可以像这样被索引。

IN: val temp = df.schema

IN: temp(0)
OUT: StructField(A,StringType,true)

IN: temp(3)
OUT: StructField(D,StructType(StructField(D1,StructType(StructField(D11,StringType,true), StructField(D12,StringType,true), StructField(D13,StringType,true)),true), StructField(D2,StringType,true), StructField(D3,StringType,true)),true)

当我尝试访问嵌套的StructType 时，会发生以下情况

IN: val temp1 = temp(3).dataType

IN: temp1(0)
OUT:
Name: Unknown Error
Message: <console>:38: error: org.apache.spark.sql.types.DataType does not take parameters
       temp1(0)
            ^
StackTrace:

我不明白的是temp 和temp1 都属于StructType 类，但temp 是可迭代的，但temp1 不是。

IN: temp.getClass
OUT: class org.apache.spark.sql.types.StructType

IN: temp1.getClass
OUT: class org.apache.spark.sql.types.StructType

我也尝试了dtypes，但在尝试访问嵌套元素时遇到了类似的问题。

IN: df.dtypes(3)(0)
OUT:
Name: Unknown Error
Message: <console>:36: error: (String, String) does not take parameters
       df.dtypes(3)(0)
                   ^
StackTrace:

那么，如何在知道子字段之前遍历架构？

【问题讨论】：

您能否更具体地说明您实际想要达到的目标？
我认为您在 sn-p 中遇到错误，因为您将 temp1 分配给 temp(3).dataType 而不是 temp(3)。 dataType 方法只返回结构的类型，而不是结构本身。
@addmeaning 我希望能够遍历架构结构。在您的最后一个答案中，我可以访问每个元素，但前提是我知道嵌套字段的确切路径。但是，我的数据集包含数百个嵌套数据字段。所以，如果我可以拥有自己的模式表示，我认为遍历数据框会更容易。

标签： scala apache-spark

【解决方案1】：

好吧，如果你想要所有嵌套列的列表，你可以编写一个这样的递归函数

鉴于：

  val schema = StructType(
    StructField("name", StringType) ::
      StructField("nameSecond", StringType) ::
      StructField("nameDouble", StringType) ::
      StructField("someStruct", StructType(
        StructField("insideS", StringType) ::
          StructField("insideD", StructType(
            StructField("inside1", StringType) :: Nil
          )) ::
          Nil
      )) ::
      Nil
  )
  val rdd = session.sparkContext.emptyRDD[Row]
  val df = session.createDataFrame(rdd, schema)

 df.printSchema()

这将产生：

root
 |-- name: string (nullable = true)
 |-- nameSecond: string (nullable = true)
 |-- nameDouble: string (nullable = true)
 |-- someStruct: struct (nullable = true)
 |    |-- insideS: string (nullable = true)
 |    |-- insideD: struct (nullable = true)
 |    |    |-- inside1: string (nullable = true)

如果你想要列的全名列表，你可以这样写：

def fullFlattenSchema(schema: StructType): Seq[String] = {
  def helper(schema: StructType, prefix: String): Seq[String] = {
    val fullName: String => String = name => if (prefix.isEmpty) name else s"$prefix.$name"
    schema.fields.flatMap {
      case StructField(name, inner: StructType, _, _) =>
        fullName(name) +: helper(inner, fullName(name))
      case StructField(name, _, _, _) => Seq(fullName(name))
    }
  }

  helper(schema, "")
}

哪个会返回：

ArraySeq(name, nameSecond, nameDouble, someStruct, someStruct.insideS, someStruct.insideD, someStruct.insideD.inside1)

【讨论】：

谢谢。这几乎是我想要的。目前，您的实现只搜索嵌套结构的第二级。您能否编辑 case 函数，以便它可以递归地执行此操作并列出所有可能的嵌套级别？
如果我有 struct 数组，你能告诉我我必须对代码进行哪些更改吗？
@AKC 我不明白，你这是什么意思？
@addmeaning 我已经用下面的模式尝试了你的代码。它跳过了 array.root 中的列 |-- name: string (nullable = true) |-- nameSecond: long (nullable = true) |-- acctNum: string (nullable = true) |-- nameDouble: array (nullable = true ) | |-- 元素：结构 (containsNull = true) | | |-- insideK: 字符串 (可为空 = true) | | |-- insideS: 字符串 (可为空 = true) | | |-- insideD: long (nullable = true) |-- inside1: long (nullable = true)
@AKC 你能在 gist.github.com 上发布你的架构并在此处发送链接吗，我看不出你的架构与原始问题有何不同

【解决方案2】：

在 Spark SQL 类型模式中，在通过它进行递归时需要担心一些复杂的数据类型，例如，StructType、ArrayType 和 MapType。编写一个完全遍历具有结构映射和映射数组的模式的函数是相当复杂的。

为了递归我遇到的大多数模式，我只需要考虑 StructType 和 ArrayType。

给定一个类似的架构：

    root
     |-- name: string (nullable = true)
     |-- nameSecond: long (nullable = true)
     |-- acctRep: string (nullable = true)
     |-- nameDouble: array (nullable = true)
     |    |-- element: struct (containsNull = true)
     |    |    |-- insideK: string (nullable = true)
     |    |    |-- insideS: string (nullable = true)
     |    |    |-- insideD: long (nullable = true)
     |-- inside1: long (nullable = true)

我会使用这样的递归函数：

    def collectAllFieldNames(schema: StructType): List[String] = 
        schema.fields.flatMap {
            case StructField(name, structType: StructType, _, _) => name :: collectAllFieldNames(structType)
            case StructField(name, ArrayType(structType: StructType, _), _, _) => name :: collectAllFieldNames(structType)
            case StructField(name, _, _, _) => name :: Nil
        }

给出结果：

    List(name, nameSecond, acctRep, nameDouble, insideK, insideS, insideK, inside1)

【讨论】：