类型不匹配;发现：所需单位：数组[org.apache.spark.sql.Dataset[org.apache.spark.sql.Row]]

Question

为什么下面的代码在return语句出现编译错误，

  def getData(queries: Array[String]): Dataset[Row] = {
    val res = spark.read.format("jdbc").jdbc(jdbcUrl, "", props).registerTempTable("")
    return res
  }

错误，

type mismatch; found : Unit required: Array[org.apache.spark.sql.Dataset[org.apache.spark.sql.Row]]

Scala 版本 2.11.11

Spark 版本 2.0.0

编辑： 实际案例

  def getDataFrames(queries: Array[String]) = {
    val jdbcResult = queries.map(query => {
      val tablename = extractTableName(query)
      if (tablename.contains("1")) {
        spark.sqlContext.read.format("jdbc").jdbc(jdbcUrl1, query, props)
      } else {
        spark.sqlContext.read.format("jdbc").jdbc(jdbcUrl2, query, props)
      }
    })
  }

在这里，我想返回迭代的组合输出，如 Array[Dataset[Row]] 或 Array[DataFrame]（但 Dataframe 在 2.0.0 中不可用作为依赖项）。上面的代码有什么魔力吗？或者我该怎么做？

Answer 1

您可以将list 中的dataframes 返回为List[Dataframe]

def getData(queries: Array[String]): List[Dataframe] = {
  val res = spark.read.format("jdbc").jdbc(jdbcUrl, "", props)
  //create multiple dataframe from your queries
  val df1 = ???
  val df2 = ???
  val list = List(df1, df2)
  //You can create a list dynamically with list of quries 
  list
}

registerTempTable 返回Unit 你最好删除registerTempTable 并返回Dataframe，并返回一个list 的数据帧。

更新：

这里是您如何使用查询列表返回数据框列表

def getDataFrames(queries: Array[String]): Array[DataFrame] = {
  val jdbcResult = queries.map(query => {
    val tablename = extractTableName(query)
    val dataframe = if (tablename.contains("1")) {
      spark.sqlContext.read.format("jdbc").jdbc("", query, prop)
    } else {
      spark.sqlContext.read.format("jdbc").jdbc("", query, prop)
    }
    dataframe
  })
  jdbcResult
}

我希望这会有所帮助！

Answer 2

从错误消息中可以清楚地看出您的函数中存在类型不匹配。 registerTempTable() api 创建一个内存表，范围为当前会话，并在 SparkSession 处于活动状态之前保持可访问性。

Check the return type of registerTempTable() api here

将您的代码更改为以下内容以删除错误消息：

def getData(queries: Array[String]): Unit = {
    val res = spark.read.format("jdbc").jdbc(jdbcUrl, "", props).registerTempTable("")

  }

更好的方法是编写如下代码：

val tempName: String = "Name_Of_Temp_View" spark.read.format("jdbc").jdbc(jdbcUrl, "", props).createOrReplaceTempView(tempName)

使用 createOrReplaceTempView() as registerTempTable() 自 Spark 2.0.0 起已弃用

根据您的要求的替代解决方案：

def getData(queries: Array[String], spark: SparkSession): Array[DataFrame] = { spark.read.format("jdbc").jdbc(jdbcUrl, "", props).createOrReplaceTempView("Name_Of_Temp_Table") val result: Array[DataFrame] = queries.map(query => spark.sql(query)) result }