想要转置 spark scala 中的列

Question

我需要使用 Scala 对 table 执行完整性检查。

我的样本 table 是：

可以使用以下代码生成它：

val df = Seq(
    (1, "", "13-05-2020", Some("CSE")),
    (2, "PQR", "14022019", None),
    (3, "XYZ", "25-01-2018", Some(""))
).toDF("ID", "NAME", "Joining Date", "Branch")

现在，我想对所有列执行 Null Check 和 Date Check。我希望输出如下：

我能够生成这样的数据框：

请帮助我将数据填充到此 DF 中。

Answer 1

据我了解，您想对数据框的所有列计算一些通用测试。

您似乎还认为空字符串 "" 类似于空值。因此，我将从用空值替换空字符串开始（仅适用于字符串类型的列）。

import org.apache.spark.sql.types.StringType
val df_with_nulls = df.select(df.schema.fields.map(c =>
         if(c.dataType == StringType)
             when(col(c.name) === "", null).otherwise(col(c.name)) as c.name
         else
             col(c.name)
) : _*)

然后，为了代码更清晰，我将为您要执行的每个检查定义一个函数。如果它们不是您想要的，请随意调整它们：

val name = (c : String) => lit(c) as s"col_list"
val null_check =
    (c : String) => sum(col(c).isNull.cast("int")) > 0 as s"Null_check"
val date_pattern = "[0-9]{2}-?[0-9]{2}-?[0-9]{4}"
val date_check =
    (c : String) => sum(col(c) rlike date_pattern cast "int") > 0 as s"Date_Check"
val distinct_values = (c : String) => collect_set(col(c)) as "Distinct_Values"

最后，您可以像这样以通用方式将所有这些应用到所有列：

df_with_nulls
    .select(array(df.columns.map(c =>
            struct(name(c), date_check(c), null_check(c), distinct_values(c))
    ) : _*) as "a")
    .select(explode('a) as "s")
    .select("s.*")
    .show(false)

同时产生您所期望的：

+------------+----------+----------+----------------------------------+
|col_list    |Date_Check|Null_check|Distinct_Values                   |
+------------+----------+----------+----------------------------------+
|ID          |false     |false     |[1, 2, 3]                         |
|NAME        |false     |true      |[PQR, XYZ]                        |
|Joining Date|true      |false     |[14022019, 13-05-2020, 25-01-2018]|
|Branch      |false     |true      |[CSE]                             |
+------------+----------+----------+----------------------------------+

我唯一的警告是不要在具有太多不同值的列上使用它。的确，由于它们都聚合到一条线上，那会很慢甚至更糟，导致 OOM 异常。

想要转置 spark scala 中的列

Want to transpose the columns in spark scala

scala

sanity-check

apache-spark