测试sparksql查询

Question

我有一个Dataframe，我想对其运行一个简单的查询，如下所示：

def runQuery(df: DataFrame, queryString: String): DataFrame = {

    df.createOrReplaceTempView("myDataFrame")
    spark.sql(queryString)

}

其中queryString可以是像这样的

"SELECT name, age FROM myDataFrame WHERE age > 30"

但我真的很想提前知道查询是否可以在不抛出异常的情况下工作。例如，如果df没有列name和age怎么办？我想写一些类似这样的东西来处理它：

def runQuery(df: DataFrame, queryString: String): DataFrame = {

    if (/*** df and queryString are compatible ***/) {
        df.createOrReplaceTempView("myDataFrame")
        spark.sql(queryString)
    } else {
        spark.createDataFrame(sc.emptyRDD[Row], df.schema)
    }

}

有没有办法在'if‘语句中检查这一点？

Answer 1

我不会太担心异常。只需用Try包装它：

import scala.util.Try
import org.apache.spark.sql.catalyst.encoders.RowEncoder

def runQuery(df: DataFrame, queryString: String): DataFrame = Try {
  df.createOrReplaceTempView("myDataFrame")
  df.sparkSession.sql(queryString)
}.getOrElse(df.sparkSession.emptyDataset(RowEncoder(df.schema)))

Answer 2

您可以使用触发spark作业检查数据帧中存在或不存在的所有列

  def runQuery(df: DataFrame, queryString: String): DataFrame = 
    if(Array("name", "age", "address").forall(df.columns.contains)) {
      df.createOrReplaceTempView("myDataFrame")
      df.sparkSession.sql(queryString)
    } else {
      df.sparkSession.emptyDataset(RowEncoder(df.schema))
    }

也可以使用df.schema来匹配数据类型