1

我想将 SQL 字符串作为用户输入，然后在执行前对其进行转换。特别是，我想修改顶层投影（select 子句），注入额外的列以供查询检索。

我希望通过使用sparkSession.experimental.extraOptimizations. 我知道我正在尝试的并不是严格意义上的优化（转换改变了 SQL 语句的语义），但 API 似乎仍然合适。但是，查询执行器似乎忽略了我的转换。

这是一个最小的例子来说明我遇到的问题。首先定义一个行案例类：

case class TestRow(a: Int, b: Int, c: Int)

然后定义一个简单地丢弃任何投影的优化规则：

object RemoveProjectOptimisationRule extends Rule[LogicalPlan] {
    def apply(plan: LogicalPlan): LogicalPlan = plan transformDown {
        case x: Project => x.child
    }
}

现在创建一个数据集，注册优化，并运行 SQL 查询：

// Create a dataset and register table.
val dataset = List(TestRow(1, 2, 3)).toDS()
val tableName: String = "testtable"
dataset.createOrReplaceTempView(tableName)

// Register "optimisation".
sparkSession.experimental.extraOptimizations =  
    Seq(RemoveProjectOptimisationRule)

// Run query.
val projected = sqlContext.sql("SELECT a FROM " + tableName + " WHERE a = 1")

// Print query result and the queryExecution object.
println("Query result:")
projected.collect.foreach(println)
println(projected.queryExecution)

这是输出：

Query result: 
[1]

== Parsed Logical Plan ==
'Project ['a]
+- 'Filter ('a = 1)
   +- 'UnresolvedRelation `testtable`

== Analyzed Logical Plan ==
a: int
Project [a#3]
+- Filter (a#3 = 1)
   +- SubqueryAlias testtable
      +- LocalRelation [a#3, b#4, c#5]

== Optimized Logical Plan ==
Filter (a#3 = 1)
+- LocalRelation [a#3, b#4, c#5]

== Physical Plan ==
*Filter (a#3 = 1)
+- LocalTableScan [a#3, b#4, c#5]

我们看到结果与原始 SQL 语句的结果相同，但未应用转换。然而，在打印逻辑和物理计划时，投影确实已被删除。我还确认（通过调试日志输出）确实正在调用转换。

关于这里发生了什么的任何建议？也许优化器只是忽略了改变语义的“优化”？

如果使用优化不是要走的路，有人可以提出替代方案吗？我真正想做的就是解析输入的 SQL 语句，对其进行转换，然后将转换后的 AST 传递给 Spark 以执行。但据我所知，执行此操作的 API 对 Sparksql包来说是私有的。可能可以使用反射，但我想避免这种情况。

任何指针将不胜感激。

4

2 回答 2

4

正如您所猜测的那样，这是行不通的，因为我们假设优化器不会改变查询的结果。

具体来说，我们缓存来自分析器的模式（并假设优化器不会更改它）。将行转换为外部格式时，我们使用此模式，因此会截断结果中的列。如果您做的不仅仅是截断（即更改数据类型），这甚至可能会崩溃。

正如您在此笔记本中看到的那样，它实际上正在产生您所期望的结果。我们计划在不久的将来的某个时候打开更多的钩子，让您可以在查询执行的其他阶段修改计划。有关详细信息，请参阅SPARK-18127。

于 2016-10-26T23:52:31.820 回答
2

Michael Armbrust 的回答证实了这种转变不应该通过优化来完成。

相反，我使用 Spark 中的内部 API 来实现我现在想要的转换。它需要 Spark 中包私有的方法。所以我们可以通过将相关逻辑放在适当的包中来访问它们而无需反射。概述：

// Must be in the spark.sql package.
package org.apache.spark.sql

object SQLTransformer {
    def apply(sparkSession: SparkSession, ...) = {

        // Get the AST.
        val ast = sparkSession.sessionState.sqlParser.parsePlan(sql)

        // Transform the AST.
        val transformedAST = ast match {
            case node: Project => // Modify any top-level projection 
            ...
        }

        // Create a dataset directly from the AST.
        Dataset.ofRows(sparkSession, transformedAST)
    }
}

请注意，这当然可能会与 Spark 的未来版本中断。

于 2016-10-27T09:43:41.367 回答