如何访问Spark UDF函数中的整行？

Question

我们使用纯SQL语法来转换数据，并具有自定义的UDF函数。示例：

UDF_FUNCTION(String, Int)

此函数可能会产生异常，我们希望为用户提供详细的错误信息。在行中，我们有关于文件和row_id的信息，这就是为什么我们要访问所提到的UDF中的整行，以提供有关file_uuid和row_id的详细错误。有人知道这件事吗？

谢谢

Answer 1

您可以使用udf函数将所有列发送到。您必须在的匿名函数中使用udf行类型作为输入参数。类似于下面的示例：

  def udf_full_row = udf { (row: Row) =>
    val your_transformed_int = (row.getAs[Int]("value as int") + 1)
    your_transformed_int
  }

  import org.apache.spark.sql.functions.{col, struct}

  val df_test : DataFrame = ???
  val cols_array = df_test.columns.map(col(_))

  df_test.withColumn("your_new_colun", udf_full_row(struct(cols_array: _*)))