斯卡拉 UDF 具有在派斯帕克中使用的多个参数

Question

我有一个用斯卡拉写的UDF，我希望能够通过Pyspark会话调用它。UDF 采用两个参数：字符串列值和第二个字符串参数。我已经能够成功地调用UDF，如果它只需要一个参数（列值）。如果需要多个参数，我很难调用UDF。以下是到目前为止我在斯卡拉和Pyspark中能够做的事情：

Scala UDF：

class SparkUDFTest() extends Serializable {
  def stringLength(columnValue: String, columnName: String): Int =
      LOG.info("Column name is: " + columnName)
      return columnValue.length
}

在Scala中使用它时，我已经能够注册和使用这个UDF：

Scala主类：

val udfInstance = new SparkUDFTest()
val stringLength = spark.sqlContext.udf.register("stringlength", udfInstance.stringLength _)
val newDF = df.withColumn("name", stringLength(col("email"), lit("email")))

以上工作成功。下面是Pyspark的尝试:

def testStringLength(colValue, colName):
  package = "com.test.example.udf.SparkUDFTest"
udfInstance = sc._jvm.java.lang.Thread.currentThread().getContextClassLoader().loadClass(testpackage).newInstance().stringLength().apply
  return Column(udfInstance(_to_seq(sc, [colValue], _to_java_column), colName))

在皮斯帕克中呼叫 UDF：

df.withColumn("email", testStringLength("email", lit("email")))

在Pyspark中执行上述操作并进行一些调整会给我以下错误：

py4j.Py4JException: Method getStringLength([]) does not exist
or
java.lang.ClassCastException: com.test.example.udf.SparkUDFTest$$anonfun$stringLength$1 cannot be cast to scala.Function1
or
TypeError: 'Column' object is not callable

我能够修改UDF以只取一个参数（列值），并且能够成功地调用它并取回一个新的Dataframe。

斯卡拉UDF级

class SparkUDFTest() extends Serializable {
  def testStringLength(): UserDefinedFunction = udf(stringLength _)
  def stringLength(columnValue: String): Int =
      LOG.info("Column name is: " + columnName)
      return columnValue.length
}

更新蟒蛇代码：

def testStringLength(colValue, colName):
  package = "com.test.example.udf.SparkUDFTest"
  udfInstance = sc._jvm.java.lang.Thread.currentThread().getContextClassLoader().loadClass(testpackage).newInstance().testStringLength().apply
  return Column(udfInstance(_to_seq(sc, [colValue], _to_java_column)))

以上工作成功。如果UDF需要额外的参数，我仍然很难调用UDF。如何在Pyspark中将第二个参数传递给UDF？

Answer 1

我能够通过使用咖喱来解决这个问题。首先将UDF注册为

def testStringLength(columnName): UserDefinedFunction = udf((colValue: String) => stringLength(colValue, colName)

称为UDF

udfInstance = sc._jvm.java.lang.Thread.currentThread().getContextClassLoader().loadClass(testpackage).newInstance().testStringLength("email").apply
df.withColumn("email", Column(udfInstance(_to_seq(sc, [col("email")], _to_java_column))))

这可以清理一下，但这就是我如何让它工作。

编辑：我使用咖喱的原因是，即使在我想作为字符串传递给UDF的第二个参数中使用“light”时，我也一直在使用“TypeError：'Col列'对象不可调用”错误。在Scala中，我没有遇到这个问题。我不确定为什么会在Pyspark中发生这种情况。这可能是由于Python解释器和Scala代码之间可能发生的一些复杂情况。仍然不清楚，但咖喱对我有用。