Spark"包. TreeNodeException"错误python"java.lang.RuntimeException：找不到pythonUDF"

Question

我在数据库上使用pySpark 2.1。

我编写了一个UDF来为pyspark数据帧的每一行生成唯一的uuid。我使用的数据帧相对较小

我知道有内置函数spark functions和zipWithinIndex（）生成行索引，但我被特别要求在这个特定项目中使用uuid。

UDF\u insert\u uuid在小数据集上运行良好，但似乎与内置的spark函数subtract冲突。

导致此错误的原因：

包裹TreeNodeException：绑定属性，树：pythonUDF0#104830

在驱动程序堆栈错误的深处，它还说：

原因：java。lang.RuntimeException：找不到pythonUDF0#104830

下面是我正在运行的代码：

import pandas
from pyspark.sql.functions import *
from pyspark.sql.types import *

import uuid

#define a python function
def insert_uuid():
  user_created_uuid = str( uuid.uuid1() )
  return user_created_uuid

#register the python function for use in dataframes
udf_insert_uuid = udf(insert_uuid, StringType())

import pandas
from pyspark.sql.functions import *
from pyspark.sql.types import *

list_of_numbers = range(1000,1050)

temp_pandasDF = pandas.DataFrame(list_of_numbers, index=None)

sparkDF = (
  spark
  .createDataFrame(temp_pandasDF, ["data_points"])
  .withColumn("labels", when( col("data_points") < 1025, "a" ).otherwise("b"))    #if "values" < 25, then "labels" = "a", else "labels" = "b"
  .repartition("labels")
)

sparkDF.createOrReplaceTempView("temp_spark_table")

#add a unique id for each row
#udf works fine in the line of code here
sparkDF = sparkDF.withColumn("id", lit( udf_insert_uuid() ))

sparkDF.show(20, False)

+-----------+------+------------------------------------+
|data_points|labels|id |
+-----------+------+------------------------------------+ 
|1029 |b |d3bb91e0-9cc8-11e7-9b70-00163e9986ba|
|1030 |b |d3bb95e6-9cc8-11e7-9b70-00163e9986ba|
|1035 |b |d3bb982a-9cc8-11e7-9b70-00163e9986ba|
|1036 |b |d3bb9a50-9cc8-11e7-9b70-00163e9986ba|
|1042 |b |d3bb9c6c-9cc8-11e7-9b70-00163e9986ba|
+-----------+------+------------------------------------+
only showing top 5 rows

list_of_numbers = range(1025,1075)

temp_pandasDF = pandas.DataFrame(list_of_numbers, index=None)

new_DF = (
  spark
  .createDataFrame(temp_pandasDF, ["data_points"])
  .withColumn("labels", when( col("data_points") < 1025, "a" ).otherwise("b"))    #if "values" < 25, then "labels" = "a", else "labels" = "b"
  .repartition("labels"))

new_DF.show(5, False)

+-----------+------+
|data_points|labels|
+-----------+------+
|1029 |b |
|1030 |b |
|1035 |b |
|1036 |b |
|1042 |b | 
+-----------+------+
only showing top 5 rows

values_not_in_new_DF = (new_DF.subtract(sparkDF.drop("id")))

display(values_not_in_new_DF
       .withColumn("id", lit( udf_insert_uuid()))   #add a column of unique uuid's
       )

Answer 1

运行脚本时，我遇到了与您相同的错误。我发现使其工作的唯一方法是传递UDF的列，而不是不传递参数：

def insert_uuid(col):
    user_created_uuid = str( uuid.uuid1() )
    return user_created_uuid
udf_insert_uuid = udf(insert_uuid, StringType())

然后在标签上调用它，例如：

values_not_in_new_DF\
    .withColumn("id", udf_insert_uuid("labels"))\
    .show()

无需使用照明