散列火花数据框的多列

Question

我需要散列 spark 数据帧的特定列。有些列具有特定的数据类型，它们基本上是标准 spark 的 DataType 类的扩展。问题是由于某种原因，在 when 情况下，某些条件无法按预期工作。

作为哈希表，我有一张地图。我们称之为 tableConfig：

val tableConfig = Map("a" -> "KEEP", "b" -> "HASH", "c" -> "KEEP", "d" -> "HASH", "e" -> "KEEP")

盐变量用于与列连接：

val salt = "abc"

散列函数如下所示：

def hashColumns(tableConfig: Map[String, String], salt: String, df: DataFrame): DataFrame = {

    val removedColumns = tableConfig.filter(_._2 == "REMOVE").keys.toList
    val hashedColumns = tableConfig.filter(_._2 == "HASH").keys.toList

    val cleanedDF = df.drop(removedColumns: _ *)

    val colTypes = cleanedDF.dtypes.toMap

    def typeFromString(s: String): DataType = s match {
      case "StringType" => StringType
      case "BooleanType" => BooleanType
      case "IntegerType" => IntegerType
      case "DateType" => DateType
      case "ShortType" => ShortType
      case "DecimalType(15,7)" => DecimalType(15,7)
      case "DecimalType(18,2)" => DecimalType(18,2)
      case "DecimalType(11,7)" => DecimalType(11,7)
      case "DecimalType(17,2)" => DecimalType(17,2)
      case "DecimalType(38,2)" => DecimalType(38,2)
      case _ => throw new TypeNotPresentException(
        "Please check types in the dataframe. The following column type is missing: ".concat(s), null
      )
    }

    val getType = colTypes.map{case (k, _) => (k, typeFromString(colTypes(k)))}

    val hashedDF = cleanedDF.columns.foldLeft(cleanedDF) {
      (memoDF, colName) =>
        memoDF.withColumn(
          colName,
          when(col(colName).isin(hashedColumns: _*) && col(colName).isNull, null).
            when(col(colName).isin(hashedColumns: _*) && col(colName).isNotNull,
              sha2(concat(col(colName), lit(salt)), 256)).otherwise(col(colName)
            )
        )
    }

    hashedDF
  }

我收到有关特定列的错误。即错误如下：

org.apache.spark.sql.AnalysisException: 由于数据类型不匹配，无法解析 '(c IN ('a', 'b', 'd', 'e'))'：参数必须相同类型但为：布尔值！=字符串；；

列名已更改。

我的搜索没有给出任何明确的解释为什么 isin 或 isNull 函数不能按预期工作。此外，我遵循特定的实现方式，并希望避免以下方法：

1) 没有 UDF。它们对我来说很痛苦。

2) 在 spark 数据框列上没有 for 循环。数据可能包含超过十亿个样本，这在性能方面会令人头疼。

Answer 1

如 cmets 中所述，第一个修复应该是删除条件 col(colName).isin(hashedColumns: _*) && col(colName).isNull，因为此检查将始终为 false。

关于这个错误，是因为col(colName)和hashedColumns的值类型不匹配。 hashedColumns 的值始终是一个字符串，因此 col(colName) 也应该是一个字符串，但在您的情况下，它似乎是一个 Boolean。

我在这里看到的最后一个问题与foldLeft 的逻辑有关。如果我理解正确，您想要实现的是浏览这些列并将sha2 应用于hashedColumns 中存在的那些。为此，您必须将代码更改为：

// 1st change: Convert each element of hashedColumns from String to Spark col
val hashArray = hashedColumns.map(lit(_))

val hashedDF = cleanedDF.columns.foldLeft(cleanedDF) {
  (memoDF, colName) =>
    memoDF.withColumn(
      colName,

      // 2nd.change: check if colName is in "a", "b", "c", "d" etc, if so apply sha2 otherwise leave the value as it is
      when(col(colName).isNotNull && array_contains(array(hashArray:_*), lit(colName)) ,
        sha2(concat(col(colName), lit(salt)), 256)
      )
    )
}

更新：

通过foldLeft 遍历所有列效率不高，并且会增加额外开销，当您有大量列时甚至更多（请参阅下面与@baitmbarek 的讨论）我添加了另一种方法而不是foldLeft 使用单选。在下一个代码中，when 仅适用于 hashedColumns。我们将列分为 nonHashedCols 和 transformCols 然后我们连接列表并将其传递给select：

val transformedCols = hashedColumns.map{ c =>
  when(col(c).isNotNull , sha2(concat(col(c), lit(salt)), 256)).as(c)
}

val nonHashedCols = (cleanedDF.columns.toSet -- hashedColumns.toSet).map(col(_)).toList

cleanedDF.select((nonHashedCols ++ transformedCols):_*)

Answer 2

（代表问题作者发布解决方案，将其从问题移至答案部分）。

@AlexandrosBiratsis 在性能和优雅的实现方面给出了非常好的解决方案。所以 hashColumns 函数将如下所示：

def hashColumns(tableConfig: Map[String, String], salt: String, df: DataFrame): DataFrame = {

    val removedCols = tableConfig.filter(_._2 == "REMOVE").keys.toList
    val hashedCols = tableConfig.filter(_._2 == "HASH").keys.toList

    val cleanedDF = df.drop(removedCols: _ *)

    val transformedCols = hashedCols.map{ c =>
      when(col(c).isNotNull , sha2(concat(col(c), lit(salt)), 256)).as(c)
    }
    val nonHashedCols = (cleanedDF.columns.toSet -- hashedCols.toSet).map(col(_)).toList
    val hashedDF = cleanedDF.select((nonHashedCols ++ transformedCols):_*)

    hashedDF
  }