将Spark逻辑计划转换为Spark Catalyst中的另一个逻辑计划

我使用Spark Catalyst来表示openCypher查询引擎ingraph的查询计划。在查询计划过程中，我想从某个逻辑计划（Plan1）转换为另一个逻辑计划（Plan2）。（我试图保持简单的问题，所以我省略了一些细节，因为这个项目完全是开源的，所以如果需要的话，我很乐意提供关于why this is necessary的更多信息。）将Spark逻辑计划转换为Spark Catalyst中的另一个逻辑计划

我能找到的最好方法是递归使用transformDown。下面是一个小例子，它将Plan1Node s转换为Plan2Node s，方法是将替换为OpA1，将OpB1替换为OpB2。

import org.apache.spark.sql.catalyst.expressions.Attribute 
import org.apache.spark.sql.catalyst.plans.logical.{LeafNode, LogicalPlan, UnaryNode} 

trait Plan1Node extends LogicalPlan 

case class OpA1() extends LeafNode with Plan1Node { 
    override def output: Seq[Attribute] = Seq() 
} 
case class OpB1(child: Plan1Node) extends UnaryNode with Plan1Node { 
    override def output: Seq[Attribute] = Seq() 
} 

trait Plan2Node extends LogicalPlan 

case class OpA2() extends LeafNode with Plan2Node { 
    override def output: Seq[Attribute] = Seq() 
} 
case class OpB2(child: Plan2Node) extends UnaryNode with Plan2Node { 
    override def output: Seq[Attribute] = Seq() 
} 

object Plan1ToPlan2 { 
    def transform(plan: Plan1Node): Plan2Node = { 
    plan.transformDown { 
     case OpA1() => OpA2() 
     case OpB1(child) => OpB2(transform(child)) 
    } 
    }.asInstanceOf[Plan2Node] 
}

这种方法完成这项工作。此代码：

val p1 = OpB1(OpA1()) 
val p2 = Plan1ToPlan2.transform(p1)

结果：

p1: OpB1 = OpB1 
+- OpA1 

p2: Plan2Node = OpB2 
+- OpA2

但是，使用asInstanceOf[Plan2Node]绝对是代码难闻的气味。我考虑使用Strategy来定义转换规则，但该类是用于从物理转换为逻辑计划。

有没有更好的方法来定义逻辑计划之间的转换？或者正在使用被认为是反模式的多个逻辑计划？

来源

2017-08-13 Gabor Szarnyas

（回答我的问题。）

更改transformDown到一个简单的模式匹配（match）和递归调用解决了此类问题：

object Plan1ToPlan2 { 
    def transform(plan: Plan1Node): Plan2Node = { 
    plan match { 
     case OpA1() => OpA2() 
     case OpB1(child) => OpB2(transform(child)) 
    } 
    } 
}

这似乎是一种安全的，（基于我对Catalyst的有限理解）惯用的解决方案。

来源

2017-08-20 10:32:54

将Spark逻辑计划转换为Spark Catalyst中的另一个逻辑计划

回答

相关问题