每次转换/操作后 Spark Dataframe 随机 UUID 更改

Posted 2023-02-19

【中文标题】每次转换/操作后 Spark Dataframe 随机 UUID 更改

【英文标题】：Spark Dataframe Random UUID changes after every transformation/action

【发布时间】：2017-08-15 02:45:00

【问题描述】：

我有一个包含生成的 UUID 的列的 Spark 数据框。 但是，每次我对数据框执行操作或转换时，它都会在每个阶段更改 UUID。

如何只生成一次 UUID，然后让 UUID 保持静态。

重现我的问题的一些示例代码如下：

def process(spark: SparkSession): Unit = 

  import spark.implicits._

  val sc = spark.sparkContext
  val sqlContext = spark.sqlContext
  sc.setLogLevel("OFF")

  // create dataframe
  val df = spark.createDataset(Array(("a", "1"), ("b", "2"), ("c", "3"))).toDF("col1", "col2")
  df.createOrReplaceTempView("df")
  df.show(false)

  // register an UDF that creates a random UUID
  val generateUUID = udf(() => UUID.randomUUID().toString)

  // generate UUID for new column
  val dfWithUuid = df.withColumn("new_uuid", generateUUID())
  dfWithUuid.show(false)
  dfWithUuid.show(false)    // uuid is different

  // new transformations also change the uuid
  val dfWithUuidWithNewCol = dfWithUuid.withColumn("col3", df.col("col2")+1)
  dfWithUuidWithNewCol.show(false)

输出是：

+----+----+
|col1|col2|
+----+----+
|a   |1   |
|b   |2   |
|c   |3   |
+----+----+

+----+----+------------------------------------+
|col1|col2|new_uuid                            |
+----+----+------------------------------------+
|a   |1   |a414e73b-24b8-4f64-8d21-f0bc56d3d290|
|b   |2   |f37935e5-0bfc-4863-b6dc-897662307e0a|
|c   |3   |e3aaf655-5a48-45fb-8ab5-22f78cdeaf26|
+----+----+------------------------------------+

+----+----+------------------------------------+
|col1|col2|new_uuid                            |
+----+----+------------------------------------+
|a   |1   |1c6597bf-f257-4e5f-be81-34a0efa0f6be|
|b   |2   |6efe4453-29a8-4b7f-9fa1-7982d2670bd6|
|c   |3   |2f7ddc1c-3e8c-4118-8e2c-8a6f526bee7e|
+----+----+------------------------------------+

+----+----+------------------------------------+----+
|col1|col2|new_uuid                            |col3|
+----+----+------------------------------------+----+
|a   |1   |00b85af8-711e-4b59-82e1-8d8e59d4c512|2.0 |
|b   |2   |94c3f2c6-9234-4fb3-b1c4-273a37171131|3.0 |
|c   |3   |1059fff2-b8f9-4cec-907d-ea181d5003a2|4.0 |
+----+----+------------------------------------+----+

请注意，每一步的 UUID 都是不同的。

【参考方案1】：

这是一种预期的行为。用户自定义函数have to be deterministic：

用户定义的函数必须是确定性的。由于优化， 重复的调用可能会被消除，或者函数甚至可能是 被调用的次数比它在查询中出现的次数多。

如果您想包含非确定性函数并保留输出，您应该将中间数据写入持久存储并读回。检查点或缓存可能在一些简单的情况下有效，但通常不会可靠。

如果上游进程是确定性的（对于初学者来说有随机播放），您可以尝试使用rand function with seed，转换为字节数组并传递给UUID.nameUUIDFromBytes。

另见：About how to add a new column to an existing DataFrame with random values in Scala

注意：SPARK-20586 引入了deterministic 标志，可以禁用某些优化，但不清楚当数据为persisted 并且发生执行器丢失时它的行为。

【讨论】：

这个问题是只有udfs才有，还是如果我们在map函数中添加列也会出现？

@abalcerek，我在地图功能中遇到了这个问题（当制作了 2 个“收集”时）

我可以确认，非确定性标志不在应用于生成带有UUID.randomUUID() 的随机 UUID 的 UDF 时的行为与宣传的一样。 UDF 仍可能被多次评估。这可能与SPARK-23599

有关

【参考方案2】：

这是一个非常古老的问题，但让人们知道什么对我有用。它可能对某人有所帮助。

您可以使用下面的 expr 函数来生成唯一的 GUID，该 GUID 在转换时不会改变。

import org.apache.spark.sql.functions._  
// create dataframe  
val df = spark.createDataset(Array(("a", "1"), ("b", "2"), ("c", "3"))).toDF("col1", "col2")   
df.createOrReplaceTempView("df")   
df.show(false)

// generate UUID for new column   
val dfWithUuid = df.withColumn("new_uuid", expr("uuid()"))
dfWithUuid.show(false)
dfWithUuid.show(false)    

// new transformations 
val dfWithUuidWithNewCol = dfWithUuid.withColumn("col3", df.col("col2")+1)
dfWithUuidWithNewCol.show(false)

输出如下：

+----+----+
|col1|col2|
+----+----+
|a   |1   |
|b   |2   |
|c   |3   |
+----+----+

+----+----+------------------------------------+
|col1|col2|new_uuid                            |
+----+----+------------------------------------+
|a   |1   |01c4ef0f-9e9b-458e-b803-5f66df1f7cee|
|b   |2   |43882a79-8e7f-4002-9740-f22bc6b20db5|
|c   |3   |64bc741a-0d7c-430d-bfe2-a4838f10acd0|
+----+----+------------------------------------+

+----+----+------------------------------------+
|col1|col2|new_uuid                            |
+----+----+------------------------------------+
|a   |1   |01c4ef0f-9e9b-458e-b803-5f66df1f7cee|
|b   |2   |43882a79-8e7f-4002-9740-f22bc6b20db5|
|c   |3   |64bc741a-0d7c-430d-bfe2-a4838f10acd0|
+----+----+------------------------------------+

+----+----+------------------------------------+----+
|col1|col2|new_uuid                            |col3|
+----+----+------------------------------------+----+
|a   |1   |01c4ef0f-9e9b-458e-b803-5f66df1f7cee|2.0 |
|b   |2   |43882a79-8e7f-4002-9740-f22bc6b20db5|3.0 |
|c   |3   |64bc741a-0d7c-430d-bfe2-a4838f10acd0|4.0 |
+----+----+------------------------------------+----+

【讨论】：

它抛出异常error: not found: value expr

我认为您在代码顶部没有 import 语句。 import org.apache.spark.sql.functions._

【参考方案3】：

我有一个 pyspark 版本：

from pyspark.sql import functions as f

pdataDF=dataDF.withColumn("uuid_column",f.expr("uuid()"))
display(pdataDF)
pdataDF.write.mode("overwrite").saveAsTable("tempUuidCheck")

【参考方案4】：

试试这个：

df.withColumn("XXXID", lit(java.util.UUID.randomUUID().toString))

它的工作原理与：

val generateUUID = udf(() => java.util.UUID.randomUUID().toString)
df.withColumn("XXXCID", generateUUID() )

我希望这会有所帮助。

帕维尔

【讨论】：

lit 将为所有数据创建相同的 uuid。