如何调用基于Json Object scala spark的方法？答案

【问题标题】：How call method based on Json Object scala spark?如何调用基于Json Object scala spark的方法？
【发布时间】：2018-05-03 09:03:55
【问题描述】：

我有如下两个函数

def method1(ip:String,r:Double,op:String)={
val data = spark.read.option("header", true).csv(ip).toDF()
val r3= data.select("c", "S").dropDuplicates("C", "S").withColumn("R", lit(r))
r3.coalesce(1).write.format("com.databricks.spark.csv").option("header", "true").save(op)
}

def method2(ip:String,op:String)={
val data = spark.read.option("header", true).csv(ip).toDF()
val r3= data.select("c", "S").dropDuplicates("C", "StockCode")
r3.coalesce(1).write.format("com.databricks.spark.csv").option("header", "true").save(op)
}

我想通过基于 Json 对象参数来调用这个方法。例如，如果我的输入 json 如下所示

{"name":"method1","ip":"Or.csv","r":1.0,"op":"oppath"}

它必须调用method1 和"Or.csv",1.0,”oppath" 作为参数，即json中的对象名表示方法名，扩孔字段为参数。

请帮助我。

【问题讨论】：

您需要解析 JSON 并在 Scala 中寻找 eval 替代方案。您必须动态执行有时会变得丑陋的代码。缩小用例范围并使用 match - case 或 if - else

标签： scala apache-spark spark-dataframe user-defined-functions scala-collections

【解决方案1】：

首先我们需要通过 spark 将 Json 读入数据帧。

val df = sqlContext.read.json("path to the json file")

应该给你数据框

scala> df.show()
+------+-------+------+---+
|    ip|   name|    op|  r|
+------+-------+------+---+
|Or.csv|method1|oppath|1.0|
+------+-------+------+---+

下一个

   scala> def method1(ip:String,r:Double,op:String)={
         | val data = spark.read.option("header", true).csv(ip).toDF()
         | val r3= data.select("c", "S").dropDuplicates("C", "S").withColumn("R", lit(r))
         | r3.coalesce(1).write.format("com.databricks.spark.csv").option("header", "true").save(op)
         | }

method1: (ip: String, r: Double, op: String)Unit

下一个

   scala> def method2(ip:String,op:String)={
         | val data = spark.read.option("header", true).csv(ip).toDF()
         | val r3= data.select("c", "S").dropDuplicates("C", "StockCode")
         | r3.coalesce(1).write.format("com.databricks.spark.csv").option("header", "true").save(op)
         | }

method2: (ip: String, op: String)Unit

下一个

scala>df.withColumn("methodCalling",when($"name" === "method1",method1(df.first().getString(1),df.first().getDouble(2),df.first().getString(3))).otherwise(when($"name" === "method2", method2(df.first().getString(1),df.first().getString(2)))))

它会根据Json Object调用method1或method2。

【讨论】：