如何为 RDD/Dataframe 中的整行创建 sha1 散列答案

【问题标题】：How to create sha1 hashing for the entire row in a RDD/Dataframe如何为 RDD/Dataframe 中的整行创建 sha1 散列
【发布时间】：2018-02-20 18:25:45
【问题描述】：

我有一个带有一个模式的数据框。现有数据框已经有 50 列。现在我想在现有数据框中添加一个新列。新列名是“hashing_id”，这个 hashing_id 的逻辑是 sha1(row)。我如何做到这一点？

我尝试了以下代码。以下这两种方法位于主要 Class 使用的特征中。这个 trait 也扩展了 Serializable

 def addHashingKey():DataFrame={
val sha1 = java.security.MessageDigest.getInstance("SHA-1")
val enCoder = new sun.misc.BASE64Encoder()
//enCoder.encode(sha1.digest(row.mkString.getBytes))
createDataFrame(df.map(row => {
        Row.fromSeq(row.toSeq ++ enCoder.encode(sha1.digest(row.mkString.getBytes)))
}), df.schema.add("hashing_id", StringType))

 }


def createDataFrame(rdd: RDD[Row], schema: StructType): DataFrame = {
sqlContext.createDataFrame(rdd, schema)
}

如何使用 rdd 实现 sha1？

有人可以帮我解决这个问题

当我运行代码时，它会抛出异常

 17/09/12 13:45:20 ERROR yarn.ApplicationMaster: User class threw exception: org.apache.spark.SparkException: Task not serializable
 org.apache.spark.SparkException: Task not serializable

 Caused by: java.io.NotSerializableException: sun.misc.BASE64Encoder
 Serialization stack:
 - object not serializable (class: sun.misc.BASE64Encoder, value:   sun.misc.BASE64Encoder@46c0813)

【问题讨论】：

标签： scala apache-spark

【解决方案1】：

你不能尝试这样的事情吗，在我刚刚运行的一些测试中它似乎对我有用：

 val newDF = sqlContext.createDataFrame(
rdd.map(x => Row(x.toSeq ++ Seq(x.toSeq.hashCode()): _*)), StructType(schema.iterator.toSeq ++ Seq(StructField("hashing_id", StringType, true))))

显然你需要将 hashCode 替换为你需要的哈希函数

编辑：使用你的 sha1 函数

在另一个类中定义你的函数

object Encoder {
  def sha1(s: Row): String = MessageDigest.getInstance("SHA-1").digest(s.mkString.getBytes()).toString
}

然后在你原来的类中你可以如下调用你的函数

   val newDF = sqlContext.createDataFrame(wordsRDD.map(x => Row(x.toSeq ++ Seq(Encoder.sha1(x)): _*)), StructType(schema.iterator.toSeq ++ Seq(StructField("hashing_id", StringType, true)))).rdd.collect()

【讨论】：

我尝试通过添加 sha1 散列并抛出 java.io.NotSerializableException
您需要将您的函数移动到顶级类。我会用一个例子来更新我的答案