在 Spark 上执行多个 SQL 查询

Question

我在 test.sql 文件中有一个 Spark SQL 查询 -

CREATE GLOBAL TEMPORARY VIEW VIEW_1 AS select a,b from abc

CREATE GLOBAL TEMPORARY VIEW VIEW_2 AS select a,b from VIEW_1

select * from VIEW_2

现在，我启动我的 spark-shell 并尝试像这样执行它 -

val sql = scala.io.Source.fromFile("test.sql").mkString
spark.sql(sql).show

这会失败并出现以下错误 -

org.apache.spark.sql.catalyst.parser.ParseException:
mismatched input '<' expecting {<EOF>, 'GROUP', 'ORDER', 'HAVING', 'LIMIT', 'OR', 'AND', 'WINDOW', 'UNION', 'EXCEPT', 'MINUS', 'INTERSECT', 'SORT', 'CLUSTER', 'DISTRIBUTE'}(line 1, pos 128)

我尝试在不同的 spark.sql 语句中一一执行这些查询，并且运行良好。问题是，我有 6-7 个查询来创建临时视图，最后我需要上一个视图的输出。有没有一种方法可以让我在单个 spark.sql 语句中运行这些 SQL。我曾研究过 Postgres SQL (Redshift)，它能够执行此类查询。在 spark sql 中，在这种情况下我将不得不维护很多文件。

Answer 1

问题在于mkString 将所有行连接在一个字符串中，无法将其正确解析为有效的 SQL 查询。

脚本文件中的每一行都应作为单独的查询执行，例如：

scala.io.Source.fromFile("test.sql").getLines()
  .filterNot(_.isEmpty)  // filter out empty lines
  .foreach(query =>
    spark.sql(query).show
  )

更新

如果查询被拆分为多行，则情况会稍微复杂一些。

我们绝对需要一个标记查询结束的标记。让它成为分号字符，就像在标准 SQL 中一样。

首先，我们从源文件中收集所有非空行：

val lines = scala.io.Source.fromFile(sqlFile).getLines().filterNot(_.isEmpty)

然后我们处理收集的行，将每个新行与前一行连接起来，如果它不以分号结尾：

val queries = lines.foldLeft(List[String]()) { case(queries, line) =>
  queries match {
    case Nil => List(line) // case for the very first line
    case init :+ last =>
      if (last.endsWith(";")) {
        // if a query ended on a previous line, we simply append the new line to the list of queries
        queries :+ line.trim
      } else {
        // the query is not terminated yet, concatenate the line with the previous one
        val queryWithNextLine = last + " " + line.trim
        init :+ queryWithNextLine
      }
  }
}