如何在 Spark Scala 中将 org.apache.spark.sql.ColumnName 转换为字符串、十进制类型？

Question

我有一个像下面这样的 JSON

{"name":"method1","parameter1":"P1name","parameter2": 1.0}

我正在加载我的 JSON 文件

val sqlContext = new org.apache.spark.sql.SQLContext(sc)
val df = sqlContext.read.json("C:/Users/test/Desktop/te.txt") 
scala> df.show()

{"name":"method1","parameter1":"P1name","parameter2": 1.0}

我有一个像下面这样的功能：

def method1(P1:String, P2:Double)={
     |  print(P1)
         print(P2)
     | }

在执行下面的代码后，我根据列名调用我的方法1，它应该执行方法1。

import org.apache.spark.sql.Column
import org.apache.spark.sql.functions._
df.withColumn("methodCalling", when($"name" === "method1", method1($"parameter1",$"parameter2")).otherwise(when($"name" === "method2", method2($"parameter1",$"parameter2")))).show(false)

但我收到波纹管错误。

val sqlContext = new org.apache.spark.sql.SQLContext(sc)
val df = sqlContext.read.json("C:/Users/test/Desktop/te.txt") 
scala> df.show()

请让我知道如何将 org.apache.spark.sql.ColumnName 数据类型转换为 String

Answer 1

当您将参数传递为

method1($"parameter1",$"parameter2")

您将列传递给函数而不是原始数据类型。因此，如果您想在函数内部应用原始数据类型操作，我建议您更改您的method1和method2as函数。并且函数必须为新列的每一行返回一个值。udfudf

import org.apache.spark.sql.functions._
def method1 = udf((P1:String, P2:Double)=>{
  print(P1)
  print(P2)
  P1+P2
})

def method2 = udf((P1:String, P2:Double)=>{
  print(P1)
  print(P2)
  P1+P2
})

那么你的withColumnapi 应该可以正常工作

df.withColumn("methodCalling", when($"name" === "method1", method1($"parameter1",$"parameter2")).otherwise(when($"name" === "method2", method2($"parameter1",$"parameter2")))).show(false)

注意：udf 函数执行数据序列化和反序列化以更改要按行处理的列数据类型，这会增加复杂性和大量内存使用。应尽可能使用火花函数