看看这个：

scala> val df = Seq(
     |     (0,"Load","employeeview", "employee.empdetails", null ),
     |     (1,"Query","employeecountview",null,"select count(*) from employeeview"),
     |     (2,"store", "employeecountview",null,null)
     |   ).toDF("id", "Operation","ViewName","DiectoryName","Query")
df: org.apache.spark.sql.DataFrame = [id: int, Operation: string ... 3 more fields]

scala> df.show()
+---+---------+-----------------+-------------------+--------------------+
| id|Operation|         ViewName|       DiectoryName|               Query|
+---+---------+-----------------+-------------------+--------------------+
|  0|     Load|     employeeview|employee.empdetails|                null|
|  1|    Query|employeecountview|               null|select count(*) f...|
|  2|    store|employeecountview|               null|                null|
+---+---------+-----------------+-------------------+--------------------+

scala> val dfcount = df.count().toInt
dfcount: Int = 3

scala> :paste
// Entering paste mode (ctrl-D to finish)

for( a <- 0 to dfcount-1){
val operation = df.filter(s"id=${a}").select("Operation").as[String].first

operation match {

case "Query" => println("matching Query") // or call a function here for Query()
case "Load" => println("matching Load") // or call a function here for Load()
case "store" => println("matching store") //
case x => println("matched " + x )

}

}

// Exiting paste mode, now interpreting.

matching Load
matching Query
matching store

scala>

编辑1：

scala> val df = Seq((3,"sam",23,9876543210L)).toDF("id","name","age","phone")
df: org.apache.spark.sql.DataFrame = [id: int, name: string ... 2 more fields]

scala> df.withColumn("json1",to_json(struct($"id",$"name",$"age",$"phone"))).show(false)
+---+----+---+----------+-------------------------------------------------+
|id |name|age|phone     |json1                                            |
+---+----+---+----------+-------------------------------------------------+
|3  |sam |23 |9876543210|{"id":3,"name":"sam","age":23,"phone":9876543210}|
+---+----+---+----------+-------------------------------------------------+

scala>

scala> df.withColumn("json1",to_json(struct(df.columns.map(col(_)):_*))).show(false)
+---+----+---+----------+-------------------------------------------------+
|id |name|age|phone     |json1                                            |
+---+----+---+----------+-------------------------------------------------+
|3  |sam |23 |9876543210|{"id":3,"name":"sam","age":23,"phone":9876543210}|
+---+----+---+----------+-------------------------------------------------+

scala>

scala> val inp=List("name","age")
cols: List[String] = List(name, age)

scala> df.withColumn("json1",to_json(struct(inp.map(col(_)):_*))).show(false)
+---+----+---+----------+-----------------------+
|id |name|age|phone     |json1                  |
+---+----+---+----------+-----------------------+
|3  |sam |23 |9876543210|{"name":"sam","age":23}|
+---+----+---+----------+-----------------------+

scala>

这是我使用数据集的解决方案。这将提供类型安全和更干净的代码。但必须以业绩为基准，变化不大。

case class EmployeeOperations(id: Int, operation: String, viewName: String,DiectoryName: String, query: String)
 val data = Seq(
    EmployeeOperations(0, "Load", "employeeview", "employee.empdetails", ""),
    EmployeeOperations(1, "Query", "employeecountview", "", "select count(*) from employeeview"),
    EmployeeOperations(2, "store", "employeecountview", "", "")
  )
  val ds: Dataset[EmployeeOperations] = spark.createDataset(data)(Encoders.product[EmployeeOperations])
  printOperation(ds).show

  def printOperation(ds: Dataset[EmployeeOperations])={
    ds.map(x => x.operation match {
      case "Query" => println("matching Query"); "Query"
      case "Load" => println("matching Load"); "Load"
      case "store" => println("matching store"); "store"
      case _ => println("Found something else") ;"Nothing"
    }
    )
  }

为了测试，我只返回了一个字符串。可以返回任何基元类型。这将返回：

scala> printOperation(ds).show
matching Load
matching Query
matching store
+-----+
|value|
+-----+
| Load|
|Query|
|store|
+-----+