Spark scala基于规则从数组列派生列

Question

我是火花和scala新手。我有一个json数组结构作为输入，类似于下面的模式。

root
|-- entity: struct (nullable = true)
|    |-- email: string (nullable = true)
|    |-- primaryAddresses: array (nullable = true)
|    |    |-- element: struct (containsNull = true)
|    |    |    |-- postalCode: string (nullable = true)
|    |    |    |-- streetAddress: struct (nullable = true)
|    |    |    |    |-- line1: string (nullable = true)

我将数组结构展平到下面的示例数据帧

+-------------+--------------------------------------+--------------------------------------+
|entity.email |entity.primaryAddresses[0].postalCode |entity.primaryAddresses[1].postalCode |....
+-------------+--------------------------------------+--------------------------------------+
|[e-mail]      |                                      |                                      |
|[e-mail]      |                                      |12345                                 |
|[e-mail]      |12345                                 |                                      |
|[e-mail]      |0                                     |0                                     |
+-------------+--------------------------------------+--------------------------------------+

我的最终目标是为数据质量度量的每一列计算存在/不存在/零计数。但在计算数据质量指标之前，我正在寻找一种方法，为每个数组列元素派生一个新列，如下所示：

• 如果特定数组元素的所有值都为空，则该元素的派生列为空
• 如果数组元素至少存在一个值，则将元素存在视为1
• 如果数组元素的所有值均为零，则我将该元素标记为零（稍后计算数据质量时，我将其校准为存在=1和零=1）

下面是一个示例中间数据帧，我试图通过为每个数组元素派生的列来实现它。原始数组元素被删除。

 
+-------------+--------------------------------------+
|entity.email |entity.primaryAddresses.postalCode    |.....
+-------------+--------------------------------------+
|[e-mail]      |                                      |
|[e-mail]      |1                                     |
|[e-mail]      |1                                     |
|[e-mail]      |0                                     |
+-------------+--------------------------------------+

输入json记录元素是动态的，可以更改。为了派生数组元素的列，我构建了一个scala映射，其中键作为列名，不带数组索引（例如：entity.primaryAddresses.postalCode），值作为数组元素的列表，以运行特定键的规则。我正在寻找一种方法来实现上述中间数据帧。

一个问题是，对于某些输入文件，在我展平数据帧后，数据帧列计数超过70k。由于记录数量预计将达到数百万，我想知道我是否应该分解每个元素以获得更好的性能，而不是将json扁平化。

欣赏任何想法。非常感谢。

Answer 1

您可以利用可以帮助您执行数据质量指标的自定义用户定义函数。

val postalUdf = udf((postalCode0: Int, postalCode1: Int) => {
        //TODO implement you logic here
    })

然后使用is创建一个新的dataframe列

df
  .withColumn("postcalCode", postalUdf(col("postalCode_0"), col("postalCode_1")))
  .show()

Answer 2

创建助手函数

使用下面的函数提取列

scala> df.printSchema
root
 |-- entity: struct (nullable = false)
 |    |-- email: string (nullable = false)
 |    |-- primaryAddresses: array (nullable = false)
 |    |    |-- element: struct (containsNull = false)
 |    |    |    |-- postalCode: string (nullable = false)
 |    |    |    |-- streetAddress: struct (nullable = false)
 |    |    |    |    |-- line1: string (nullable = false)

import org.apache.spark.sql.{DataFrame, SparkSession}
import org.apache.spark.sql.functions._
import org.apache.spark.sql.types._
import scala.annotation.tailrec
import scala.util.Try

implicit class DFHelpers(df: DataFrame) {
    def columns = {
      val dfColumns = df.columns.map(_.toLowerCase)
      df.schema.fields.flatMap { data =>
        data match {
          case column if column.dataType.isInstanceOf[StructType] => {
            column.dataType.asInstanceOf[StructType].fields.map { field =>
              val columnName = column.name
              val fieldName = field.name
              col(s"${columnName}.${fieldName}").as(s"${columnName}_${fieldName}")
            }.toList
          }
          case column => List(col(s"${column.name}"))
        }
      }
    }

    def flatten: DataFrame = {
      val empty = df.schema.filter(_.dataType.isInstanceOf[StructType]).isEmpty
      empty match {
        case false =>
          df.select(columns: _*).flatten
        case _ => df
      }
    }
    def explodeColumns = {
      @tailrec
      def columns(cdf: DataFrame):DataFrame = cdf.schema.fields.filter(_.dataType.typeName == "array") match {
        case c if !c.isEmpty => columns(c.foldLeft(cdf)((dfa,field) => {
          dfa.withColumn(field.name,explode_outer(col(s"${field.name}"))).flatten
        }))
        case _ => cdf
      }
      columns(df.flatten)
    }
}

scala> df.explodeColumns.printSchema
root
 |-- entity_email: string (nullable = false)
 |-- entity_primaryAddresses_postalCode: string (nullable = true)
 |-- entity_primaryAddresses_streetAddress_line1: string (nullable = true)