Scala 中的结构类型扁平化

Question

我正在尝试从 Spark 数据框架中的结构类型创建一个列表。架构看起来像这样

root
|
|-- plotList: array (nullable = true)
|    |-- element: string (containsNull = true)
|-- plot: struct (nullable = true)
|    |-- test: struct (nullable = true)
|    |    |-- body: string (nullable = true)
|    |    |-- colorPair: struct (nullable = true)
|    |    |    |-- background: string (nullable = true)
|    |    |    |-- foreground: string (nullable = true)
|    |    |-- eta: struct (nullable = true)
|    |    |    |-- etaText: string (nullable = true)
|    |    |    |-- etaType: string (nullable = true)
|    |    |    |-- etaValue: string (nullable = true)
|    |    |-- headline: string (nullable = true)
|    |    |-- plotType: string (nullable = true)
|    |    |-- priority: long (nullable = true)
|    |    |-- plotCategory: string (nullable = true)
|    |    |-- productType: string (nullable = true)
|    |    |-- theme: string (nullable = true)
|    |-- temp: struct (nullable = true)
|    |    |-- body: string (nullable = true)
|    |    |-- colorPair: struct (nullable = true)
|    |    |    |-- background: string (nullable = true)
|    |    |    |-- foreground: string (nullable = true)
|    |    |-- eta: struct (nullable = true)
|    |    |    |-- etaText: string (nullable = true)
|    |    |    |-- etaType: string (nullable = true)
|    |    |    |-- etaValue: string (nullable = true)
|    |    |-- headline: string (nullable = true)
|    |    |-- logo: string (nullable = true)
|    |    |-- plotType: string (nullable = true)
|    |    |-- priority: long (nullable = true)
|    |    |-- plotCategory: string (nullable = true)
|    |    |-- plotType: string (nullable = true)
|    |    |-- theme: string (nullable = true)

我正在尝试编写一个 UDF，它可以将 plot 列转换为可以在下一次迭代中分解的元素列表。情节线上的东西 - > [test,temp] 我可以从 test 和 temp 中选择一些特定的列。非常感谢任何正确方向的指示。我尝试了 UDF 的多种变体，但它们似乎都不起作用。

编辑：

我想从绘图列的子列创建一个扁平结构。我正在考虑为此使用案例类。类似的东西

case class ColorPair(back:String, fore:String)
case class Eta(EtaText: String, EtaType: String, EtaValue: String)
case class Plot(body:String, colorPair: ColorPair, eta: Eta, headline: String, plotType: String, priority: String, plotCategory: String, plotType: String, theme: String)

所以，基本上在这结束时，我期待像List(Plot) 这样的东西，然后我可以在后续步骤中explode。因为 explode 不能直接在 Struct Types 上工作，所以我必须进行这个转换。在 python 世界中，我很容易将此列作为字典阅读，但在 Scala 中不存在类似的内容（据我所知）。

Answer 1

如果我理解正确，那么您正在寻找一种方法来遍历您的架构，当找到 colorPair 或 eta 时，返回这些字段：

plot.test.colorPair
plot.test.eta
plot.temp.colorPair
plot.temp.eta

为了为您的案例生成数据（模式），我编写了以下代码：

  case class Eta(etaText: String, etaType: String, etaValue: String)
  case class ColorPair(background: String, foreground: String)
  case class Test(body: String, colorPair: ColorPair, eta: Eta, headline: String, plotType: String, priority: Long, plotCategory: String, productType: String, theme: String)
  case class Temp(body: String, colorPair: ColorPair, eta: Eta ,headline: String, logo: String, plotType: String, priority: Long, plotCategory: String, productType: String, theme: String)
  case class Plot(test: Test, temp: Temp)
  case class Root(plotList: Array[String], plot: Plot)

  def getSchema(): StructType ={
    import org.apache.spark.sql.types._
    import org.apache.spark.sql.catalyst.ScalaReflection
    val schema = ScalaReflection.schemaFor[Root].dataType.asInstanceOf[StructType]

    schema.printTreeString()
    schema
  }

这将有输出：

root
 |-- plotList: array (nullable = true)
 |    |-- element: string (containsNull = true)
 |-- plot: struct (nullable = true)
 |    |-- test: struct (nullable = true)
 |    |    |-- body: string (nullable = true)
 |    |    |-- colorPair: struct (nullable = true)
 |    |    |    |-- background: string (nullable = true)
 |    |    |    |-- foreground: string (nullable = true)
 |    |    |-- eta: struct (nullable = true)
 |    |    |    |-- etaText: string (nullable = true)
 |    |    |    |-- etaType: string (nullable = true)
 |    |    |    |-- etaValue: string (nullable = true)
 |    |    |-- headline: string (nullable = true)
 |    |    |-- plotType: string (nullable = true)
 |    |    |-- priority: long (nullable = false)
 |    |    |-- plotCategory: string (nullable = true)
 |    |    |-- productType: string (nullable = true)
 |    |    |-- theme: string (nullable = true)
 |    |-- temp: struct (nullable = true)
 |    |    |-- body: string (nullable = true)
 |    |    |-- colorPair: struct (nullable = true)
 |    |    |    |-- background: string (nullable = true)
 |    |    |    |-- foreground: string (nullable = true)
 |    |    |-- eta: struct (nullable = true)
 |    |    |    |-- etaText: string (nullable = true)
 |    |    |    |-- etaType: string (nullable = true)
 |    |    |    |-- etaValue: string (nullable = true)
 |    |    |-- headline: string (nullable = true)
 |    |    |-- logo: string (nullable = true)
 |    |    |-- plotType: string (nullable = true)
 |    |    |-- priority: long (nullable = false)
 |    |    |-- plotCategory: string (nullable = true)
 |    |    |-- productType: string (nullable = true)
 |    |    |-- theme: string (nullable = true)

最后，下一个代码应该展平所需的字段：

def flattenSchema(schema: StructType, targetFields: List[String], prefix: String = null): Array[String]=
  {
    import org.apache.spark.sql.types._
    schema.fields.flatMap(f => {
      val colName = if (prefix == null) f.name else (prefix + "." + f.name)

      f.dataType match {
        case st : StructType =>
          val found = st.filter(s => targetFields.contains(s.name))

          if(found.isEmpty) {
            flattenSchema(st, targetFields, colName)
          }
          else
            found.flatMap(sf => {
              val st = sf.dataType.asInstanceOf[StructType]
              st.map(st => s"${colName}.${sf.name}.${st.name}")
            })

        case _ => Array[String]()
      }
    })
  }

上面的代码扫描架构以查找存在于targetFields 列表中的字段，然后使用flatMap 检索这些字段的架构。

这应该是输出：

plot.test.colorPair.background
plot.test.colorPair.foreground
plot.test.eta.etaText
plot.test.eta.etaType
plot.test.eta.etaValue
plot.temp.colorPair.background
plot.temp.colorPair.foreground
plot.temp.eta.etaText
plot.temp.eta.etaType
plot.temp.eta.etaValue