代码之家 › 专栏 › 技术社区 › Sai

如何转换组织。阿帕奇。火花sql。ColumnName到string,在Spark Scala中输入十进制?

user-defined-functions apache-spark-sql apache-spark scala

4

Sai · 技术社区 · 8 年前

我有一个类似下面的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|parameter1|parameter2|
+-------+----------+----------+
|method1|    P1name|    1.0   |
+-------+----------+----------+

我的函数如下:

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

在执行下面的代码后,我基于列名调用method1,它应该执行method1。

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)

但是我犯了一个错误。

<console>:63: error: type mismatch;
 found   : org.apache.spark.sql.ColumnName
 required: String

请让我知道如何转换组织。阿帕奇。火花sql。ColumnName数据类型到字符串

3 回复 | 直到 7 年前

1

5

Ramesh Maharjan 8 年前

当您将参数传递为

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

您正在向函数传递列,而不是原始数据类型。所以,我建议你改变你的 method1 和 method2 像 udf 函数,如果要应用 原始数据类型操作 内部功能。和 自定义项 函数必须为新列的每一行返回一个值。

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
})

然后你的 withColumn api应正常工作

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

注意:udf函数执行数据序列化和反序列化,以更改要按行处理的列数据类型,这将增加复杂性和大量内存使用。 spark functions 应尽可能多地使用

2

3

Learner 8 年前

您可以这样尝试:

scala> def method1(P1:String, P2:Double): Int = {
     |   println(P1)
     |   println(P2)
     |   0
     | }

scala> def method2(P1:String, P2:Double): Int = {
     |   println(P1)
     |   println(P2)
     |   1
     | }

df.withColumn("methodCalling", when($"name" === "method1", method1(df.select($"parameter1").map(_.getString(0)).collect.head,df.select($"parameter2").map(_.getDouble(0)).collect.head))
  .otherwise(when($"name" === "method2", method2(df.select($"parameter1").map(_.getString(0)).collect.head,df.select($"parameter2").map(_.getDouble(0)).collect.head)))).show

//output

P1name
1.0
+-------+----------+----------+-------------+
|   name|parameter1|parameter2|methodCalling|
+-------+----------+----------+-------------+
|method1|    P1name|       1.0|            0|
+-------+----------+----------+-------------+

您必须从方法中返回一些内容,否则它将重新运行单元,并在打印结果后给出错误:

java.lang.RuntimeException: Unsupported literal type class scala.runtime.BoxedUnit ()
  at org.apache.spark.sql.catalyst.expressions.Literal$.apply(literals.scala:75)
  at org.apache.spark.sql.functions$.lit(functions.scala:101)
  at org.apache.spark.sql.functions$.when(functions.scala:1245)
  ... 50 elided

谢谢

3

0

philantrovert 8 年前

我认为您只需要阅读JSON并基于该调用调用方法。

由于您已经创建了数据帧,因此可以执行以下操作:

df.map( row => (row.getString(0), row.getString(1) , row.getDouble(2)   ) ).collect
  .foreach { x =>
      x._1.trim.toLowerCase match {
          case "method1" => method1(x._2, x._3) 
        //case "method2" => method2(x._2, x._3)
        //case _ => methodn(x._2, x._3)
      }
   }
// Output : P1name1.0
// Because you used `print` and not `println` ;)