scala - 在现有数据框上使用 foldLeft 输出数据框

Question

我有一个 spark 数据框，我想使用 foldLeft(或任何其他方法)将聚合函数应用于每一列。应用于列的聚合函数将取决于列的数据类型。

请注意，由于我将使用大型数据框，因此我不想使用 .collect() 或任何将大量内容写入驱动程序的东西。

初始数据框如下所示:

+----------------+-----------------+------------------+
| id(StringType) | lat(DoubleType) | long(DoubleType) |
+----------------+-----------------+------------------+
| ID1            | 10.2            | 20.1             |
| ID2            | 11.1            | 50.1             |
| ID3            | null            | null             |
+----------------+-----------------+------------------+

对于此示例，我想计算所有数据类型的空值计数，只计算 DoubleType 的平均值，并且只计算 StringType 列的基数。

这是我实现 foldLeft 的框架代码，但它可能不是正确的方法。

def ourMethod(df: DataFrame): DataFrame = {
  val columns = df.schema.fields
  val initDf = spark.emptyDataFrame
  columns.foldLeft(...)((tempDf, column) => {
    column match {
      case StructField(name, dataType, _, _) => {
        dataType match {
          case StringType => ... //something like df.select("column").approx_count_distinct(), though writes in driver.
          case DoubleType => ... //something like df.agg(avg(column))
        }
      }
    }
  })
}

预期的输出如下所示:

+----------+---------+-------+-------------+
| col_name | is_null | mean  | cardinality |
+----------+---------+-------+-------------+
| id       |       0 | null  | 3           |
| lat      |       1 | 10.65 | null        |
| long     |       1 | 35.1  | null        |
+----------+---------+-------+-------------+

Answer 1

不确定 foldLeft 在这里是否有帮助，但它绝对可行。给定数据框

val df =
  Seq(("ID1", Some(10.2), Some(20.1)),
      ("ID2", Some(11.1), Some(50.1)),
      ("ID3", None, None))
    .toDF("id", "lat", "lon")

我们可以采取几种方法。

1. 以编程方式创建聚合函数。很简单

val aggs = df.schema.fields.flatMap {
  case StructField(name, DoubleType, _, _) =>
    Seq(max(col(name).isNull) as s"${name}_is_null",
        mean(col(name)) as s"${name}_mean")
  case StructField(name, StringType, _, _) =>
    Seq(max(col(name).isNull) as s"${name}_is_null",
        max(length(col(name))) as s"${name}_cardinality")
}

df.agg(aggs.head, aggs.tail: _*).show()

但是，输出将在一行中而不是所要求的。当然，这一行可以是例如收集到驱动程序并修改或平面映射到所需的格式。这是未处理的输出:

+----------+--------------+-----------+------------------+-----------+--------+
|id_is_null|id_cardinality|lat_is_null|          lat_mean|lon_is_null|lon_mean|
+----------+--------------+-----------+------------------+-----------+--------+
|     false|             3|       true|10.649999999999999|       true|    35.1|
+----------+--------------+-----------+------------------+-----------+--------+

1. 将行展开为一种格式，其中行名称是可用于分组的列，并且可能的值包含在可为空的字段中。这是因为聚合中省略了 null 值

case class FlatRow(name: String, d: Option[Double], s: Option[String])

df.flatMap { row: Row =>
    row.schema.fields.zipWithIndex.map {
      case (StructField(name, DoubleType, _, _), index) =>
        FlatRow(name,
                if (row.isNullAt(index)) None
                else Some(row.getDouble(index)),
                None)
      case (StructField(name, StringType, _, _), index) =>
        FlatRow(name,
                None,
                if (row.isNullAt(index)) None
                else Some(row.getString(index)))
    }
  }
  .groupBy($"name")
  .agg(max($"d".isNull && $"s".isNull) as "is_null",
       mean($"d") as "mean",
       max(length($"s")) as "cardinality")
  .show()

多一点代码，但它输出要求的格式:

+----+-------+------------------+-----------+
|name|is_null|              mean|cardinality|
+----+-------+------------------+-----------+
| lat|   true|10.649999999999999|       null|
| lon|   true|              35.1|       null|
|  id|  false|              null|          3|
+----+-------+------------------+-----------+