Sparkathon @ Warsaw 2016-10-12

Main.scala

import org.apache.spark.sql.SparkSession

object Main extends App {
  val spark = SparkSession.builder().master("local[*]").getOrCreate()
  spark.experimental.extraOptimizations = Seq(MyRule)
  import spark.implicits._

  val numbersDS = Seq(0,1,2,3,4,5,6,7,8,9).toDS()
  println(numbersDS.map(_+2).explain(true))

  spark.stop()
}

MyRule.scala

import org.apache.spark.sql.catalyst.expressions._
import org.apache.spark.sql.catalyst.plans.logical._
import org.apache.spark.sql.catalyst.rules._
import org.apache.spark.sql.types._

object MyRule extends Rule[LogicalPlan] with PredicateHelper {
  def apply(plan: LogicalPlan): LogicalPlan = {
    plan match {
      case Filter(Literal(true, BooleanType), child) => {
        println("I found a filter!!! yayaayaya!")
        child
      }
      case _ => {
        println("Anything else :P")
        plan
      }
    }

  }
}