如何在不执行的情况下验证Spark SQL表达式？

Question

我想验证一下spark-sql查询在语法上是否正确，而不是在集群上实际运行该查询。

实际的用例是，我正在尝试开发一个用户界面，它接受用户输入spark-sql查询，并且我应该能够验证所提供的查询在语法上是否正确。此外，如果在解析查询之后，我可以就spark最佳实践给出关于查询的任何建议。

Answer 1

SparkSqlParser

Spark SQL使用SparkSqlParser 作为Spark SQL表达式的解析器。

您可以使用SparkSession (和SessionState)访问SparkSqlParser，如下所示：

val spark: SparkSession = ...
val parser = spark.sessionState.sqlParser

scala> parser.parseExpression("select * from table")
res1: org.apache.spark.sql.catalyst.expressions.Expression = ('select * 'from) AS table#0

提示：为INFO记录器启用org.apache.spark.sql.execution.SparkSqlParser日志记录级别，以查看内部发生了什么。

SparkSession.sql方法

单凭这一点并不能为您提供最好的方法来抵御错误的SQL表达式，因此认为sql方法更合适。

SQL (sqlText: String)：DataFrame使用Spark执行SQL查询，并将结果作为DataFrame返回。用于SQL解析的方言可以使用“spark.sql.Dialect”进行配置。

请参阅下面的操作。

scala> parser.parseExpression("hello world")
res5: org.apache.spark.sql.catalyst.expressions.Expression = 'hello AS world#2

scala> spark.sql("hello world")
org.apache.spark.sql.catalyst.parser.ParseException:
mismatched input 'hello' expecting {'(', 'SELECT', 'FROM', 'ADD', 'DESC', 'WITH', 'VALUES', 'CREATE', 'TABLE', 'INSERT', 'DELETE', 'DESCRIBE', 'EXPLAIN', 'SHOW', 'USE', 'DROP', 'ALTER', 'MAP', 'SET', 'RESET', 'START', 'COMMIT', 'ROLLBACK', 'REDUCE', 'REFRESH', 'CLEAR', 'CACHE', 'UNCACHE', 'DFS', 'TRUNCATE', 'ANALYZE', 'LIST', 'REVOKE', 'GRANT', 'LOCK', 'UNLOCK', 'MSCK', 'EXPORT', 'IMPORT', 'LOAD'}(line 1, pos 0)

== SQL ==
hello world
^^^

  at org.apache.spark.sql.catalyst.parser.ParseException.withCommand(ParseDriver.scala:217)
  at org.apache.spark.sql.catalyst.parser.AbstractSqlParser.parse(ParseDriver.scala:114)
  at org.apache.spark.sql.execution.SparkSqlParser.parse(SparkSqlParser.scala:48)
  at org.apache.spark.sql.catalyst.parser.AbstractSqlParser.parsePlan(ParseDriver.scala:68)
  at org.apache.spark.sql.SparkSession.sql(SparkSession.scala:638)
  ... 49 elided

Answer 2

根据@JacekLaskowski的回答，我发现SparkSqlParser给了我各种各样的错误，这些错误并不是真正的语法错误。

因此，我同意他的观点，并建议简单地将其放入SparkSession.sql中，这很好用。下面是我的方法：

  /**
   * Validates a Spark SQL statement by trying to execute it and checking
   * if there are no syntax-related exceptions.
   */
  def validate(sqlStatement: String): Unit = {
    val spark = SparkSession.builder
      .master("local")
      .getOrCreate()
    try {
      spark.sql(sqlStatement)
    } catch {
      case ex: ParseException => throw new MyCustomException("Invalid Spark SQL", ex)
      case _: AnalysisException => // Syntax was correct
    }
  }

Answer 3

对于PySpark，您可以使用以下命令

from pyspark.sql import SparkSession
from pyspark.sql.utils import ParseException


spark_session = SparkSession \ 
    .builder \
    .master("local[1]") \
    .appName('test') \
    .getOrCreate()

try:
    parser = spark_session._jsparkSession.sessionState().sqlParser()
    parser.parseExpression(query_string)
    print('Validation successful')
except ParseException as e:
    # do something