文章详情

短信预约-IT技能 免费直播动态提醒

请输入下面的图形验证码

提交验证

短信预约提醒成功

通过扩展 Spark SQL ,打造自己的大数据分析引擎

2024-12-03 04:20

关注

Spark SQL 的 Catalyst ,这部分真的很有意思,值得去仔细研究一番,今天先来说说Spark的一些扩展机制吧,上一次写Spark,对其SQL的解析进行了一定的魔改,今天我们按套路来,使用砖厂为我们提供的机制,来扩展Spark...

首先我们先来了解一下 Spark SQL 的整体执行流程,输入的查询先被解析成未关联元数据的逻辑计划,然后根据元数据和解析规则,生成逻辑计划,再经过优化规则,形成优化过的逻辑计划(RBO),将逻辑计划转换成物理计划在经过代价模型(CBO),输出真正的物理执行计划。

我们今天举三个扩展的例子,来进行说明。

扩展解析器

这个例子,我们扩展解析引擎,我们对输入的SQL,禁止泛查询即不许使用select *来做查询,以下是解析的代。

  1. package wang.datahub.parser 
  2.  
  3. import org.apache.spark.sql.catalyst.analysis.UnresolvedStar 
  4. import org.apache.spark.sql.catalyst.expressions.Expression 
  5. import org.apache.spark.sql.catalyst.parser.ParserInterface 
  6. import org.apache.spark.sql.catalyst.plans.logical.{LogicalPlan, Project} 
  7. import org.apache.spark.sql.catalyst.{FunctionIdentifier, TableIdentifier} 
  8. import org.apache.spark.sql.types.{DataType, StructType} 
  9.  
  10. class MyParser(parser: ParserInterface) extends ParserInterface { 
  11.   
  12.  override def parsePlan(sqlText: String): LogicalPlan = { 
  13.    val logicalPlan = parser.parsePlan(sqlText) 
  14.    logicalPlan transform { 
  15.      case project @ Project(projectList, _) => 
  16.        projectList.foreach { 
  17.          name => 
  18.            if (name.isInstanceOf[UnresolvedStar]) { 
  19.              throw new RuntimeException("You must specify your project column set," + 
  20.                " * is not allowed."
  21.           } 
  22.       } 
  23.        project 
  24.   } 
  25.    logicalPlan 
  26.  
  27.   
  28.  override def parseExpression(sqlText: String): Expression = parser.parseExpression(sqlText) 
  29.  
  30.   
  31.  override def parseTableIdentifier(sqlText: String): TableIdentifier = 
  32.    parser.parseTableIdentifier(sqlText) 
  33.  
  34.   
  35.  override def parseFunctionIdentifier(sqlText: String): FunctionIdentifier = 
  36.    parser.parseFunctionIdentifier(sqlText) 
  37.  
  38.   
  39.  override def parseTableSchema(sqlText: String): StructType = 
  40.    parser.parseTableSchema(sqlText) 
  41.  
  42.   
  43.  override def parseDataType(sqlText: String): DataType = parser.parseDataType(sqlText) 

接下来,我们测试一下

  1. package wang.datahub.parser 
  2.  
  3. import org.apache.spark.sql.{SparkSession, SparkSessionExtensions} 
  4. import org.apache.spark.sql.catalyst.parser.ParserInterface 
  5.  
  6. object MyParserApp { 
  7.  def main(args: Array[String]): Unit = { 
  8.    System.setProperty("hadoop.home.dir","E:\\devlop\\envs\\hadoop-common-2.2.0-bin-master"); 
  9.    type ParserBuilder = (SparkSession, ParserInterface) => ParserInterface 
  10.    type ExtensionsBuilder = SparkSessionExtensions => Unit 
  11.    val parserBuilder: ParserBuilder = (_, parser) => new MyParser(parser) 
  12.    val extBuilder: ExtensionsBuilder = { e => e.injectParser(parserBuilder)} 
  13.    val spark =  SparkSession 
  14.     .builder() 
  15.     .appName("Spark SQL basic example"
  16.     .config("spark.master""local[*]"
  17.     .withExtensions(extBuilder) 
  18.     .getOrCreate() 
  19.  
  20.    spark.sparkContext.setLogLevel("ERROR"
  21.  
  22.    import spark.implicits._ 
  23.  
  24.    val df = Seq( 
  25.     ( "First Value",1, java.sql.Date.valueOf("2010-01-01")), 
  26.     ( "First Value",4, java.sql.Date.valueOf("2010-01-01")), 
  27.     ("Second Value",2,  java.sql.Date.valueOf("2010-02-01")), 
  28.     ("Second Value",9,  java.sql.Date.valueOf("2010-02-01")) 
  29.   ).toDF("name""score""date_column"
  30.    df.createTempView("p"
  31.  
  32.    //   val df = spark.read.json("examples/src/main/resources/people.json"
  33.    //   df.toDF().write.saveAsTable("person"
  34.    //,javg(score) 
  35.  
  36.    // custom parser 
  37.    //   spark.sql("select * from p ").show 
  38.  
  39.    spark.sql("select * from p").show() 

下面是执行结果,符合我们的预期。

扩展优化器

接下来,我们来扩展优化器,砖厂提供了很多默认的RBO,这里可以方便的构建我们自己的优化规则,本例中我们构建一套比较奇怪的规则,而且是完全不等价的,这里只是为了说明。

针对字段+0的操作,规则如下:

  1. 如果0出现在+左边,则直接将字段变成右表达式,即 0+nr 等效为 nr
  2. 如果0出现在+右边,则将0变成3,即 nr+0 变成 nr+3
  3. 如果没出现0,则表达式不变

下面是代码:

  1. package wang.datahub.optimizer 
  2.  
  3. import org.apache.spark.sql.SparkSession 
  4. import org.apache.spark.sql.catalyst.expressions.{Add, Expression, Literal} 
  5. import org.apache.spark.sql.catalyst.plans.logical.LogicalPlan 
  6. import org.apache.spark.sql.catalyst.rules.Rule 
  7.  
  8. object MyOptimizer extends Rule[LogicalPlan] { 
  9.  
  10.  def apply(logicalPlan: LogicalPlan): LogicalPlan = { 
  11.    logicalPlan.transformAllExpressions { 
  12.      case Add(leftright) => { 
  13.        println("this this my add optimizer"
  14.        if (isStaticAdd(left)) { 
  15.          right 
  16.       } else if (isStaticAdd(right)) { 
  17.          Add(left, Literal(3L)) 
  18.       } else { 
  19.          Add(leftright
  20.       } 
  21.     } 
  22.   } 
  23.  
  24.  private def isStaticAdd(expression: Expression): Boolean = { 
  25.    expression.isInstanceOf[Literal] && expression.asInstanceOf[Literal].toString == "0" 
  26.  
  27.  def main(args: Array[String]): Unit = { 
  28.    System.setProperty("hadoop.home.dir","E:\\devlop\\envs\\hadoop-common-2.2.0-bin-master"); 
  29.    val testSparkSession: SparkSession = SparkSession.builder().appName("Extra optimization rules"
  30.     .master("local[*]"
  31.     .withExtensions(extensions => { 
  32.        extensions.injectOptimizerRule(session => MyOptimizer) 
  33.     }) 
  34.     .getOrCreate() 
  35.  
  36.    testSparkSession.sparkContext.setLogLevel("ERROR"
  37.  
  38.    import testSparkSession.implicits._ 
  39.    testSparkSession.experimental.extraOptimizations = Seq() 
  40.    Seq(-1, -2, -3).toDF("nr").write.mode("overwrite").json("./test_nrs"
  41. //   val optimizedResult = testSparkSession.read.json("./test_nrs").selectExpr("nr + 0"
  42.    testSparkSession.read.json("./test_nrs").createTempView("p"
  43.  
  44.    var sql = "select nr+0 from p"
  45.    var t = testSparkSession.sql(sql) 
  46.    println(t.queryExecution.optimizedPlan) 
  47.    println(sql) 
  48.    t.show() 
  49.  
  50.    sql = "select 0+nr from p"
  51.    var  u = testSparkSession.sql(sql) 
  52.    println(u.queryExecution.optimizedPlan) 
  53.    println(sql) 
  54.    u.show() 
  55.  
  56.    sql = "select nr+8 from p"
  57.    var  v = testSparkSession.sql(sql) 
  58.    println(v.queryExecution.optimizedPlan) 
  59.    println(sql) 
  60.    v.show() 
  61. //   println(optimizedResult.queryExecution.optimizedPlan.toString() ) 
  62. //   optimizedResult.collect().map(row => row.getAs[Long]("(nr + 0)")) 
  63.    Thread.sleep(1000000) 
  64.  

执行如下

  1. this this my add optimizer 
  2. this this my add optimizer 
  3. this this my add optimizer 
  4. Project [(nr#12L + 3) AS (nr + CAST(0 AS BIGINT))#14L] 
  5. +- Relation[nr#12L] json 
  6.  
  7. select nr+0 from p 
  8. this this my add optimizer 
  9. this this my add optimizer 
  10. this this my add optimizer 
  11. +------------------------+ 
  12. |(nr + CAST(0 AS BIGINT))| 
  13. +------------------------+ 
  14. |                       2| 
  15. |                       1| 
  16. |                       0| 
  17. +------------------------+ 
  18.  
  19. this this my add optimizer 
  20. Project [nr#12L AS (CAST(0 AS BIGINT) + nr)#21L] 
  21. +- Relation[nr#12L] json 
  22.  
  23. select 0+nr from p 
  24. this this my add optimizer 
  25. +------------------------+ 
  26. |(CAST(0 AS BIGINT) + nr)| 
  27. +------------------------+ 
  28. |                     -1| 
  29. |                     -2| 
  30. |                     -3| 
  31. +------------------------+ 
  32.  
  33. this this my add optimizer 
  34. this this my add optimizer 
  35. this this my add optimizer 
  36. Project [(nr#12L + 8) AS (nr + CAST(8 AS BIGINT))#28L] 
  37. +- Relation[nr#12L] json 
  38.  
  39. select nr+8 from p 
  40. this this my add optimizer 
  41. this this my add optimizer 
  42. this this my add optimizer 
  43. +------------------------+ 
  44. |(nr + CAST(8 AS BIGINT))| 
  45. +------------------------+ 
  46. |                       7| 
  47. |                       6| 
  48. |                       5| 
  49. +------------------------+ 

扩展策略

SparkStrategies包含了一系列特定的Strategies,这些Strategies是继承自QueryPlanner中定义的Strategy,它定义接受一个Logical Plan,生成一系列的Physical Plan

通过Strategies把逻辑计划转换成可以具体执行的物理计划,代码如下

  1. package wang.datahub.strategy 
  2.  
  3. import org.apache.spark.sql.{SparkSession, Strategy} 
  4. import org.apache.spark.sql.catalyst.plans.logical.LogicalPlan 
  5. import org.apache.spark.sql.execution.SparkPlan 
  6.  
  7. object MyStrategy extends Strategy { 
  8.  def apply(plan: LogicalPlan): Seq[SparkPlan] = { 
  9.    println("Hello world!"
  10.    Nil 
  11.  
  12.  def main(args: Array[String]): Unit = { 
  13.    System.setProperty("hadoop.home.dir","E:\\devlop\\envs\\hadoop-common-2.2.0-bin-master"); 
  14.    val spark = SparkSession.builder().master("local").getOrCreate() 
  15.  
  16.    spark.experimental.extraStrategies = Seq(MyStrategy) 
  17.    val q = spark.catalog.listTables.filter(t => t.name == "six"
  18.    q.explain(true
  19.    spark.stop() 

执行效果

好了,扩展部分就先介绍到这,接下来我计划可能会简单说说RBO和CBO,结合之前做过的一个小功能,一条SQL的查询时间预估。

本文转载自微信公众号「麒思妙想」,可以通过以下二维码关注。转载本文请联系麒思妙想公众号。

 

来源:麒思妙想内容投诉

免责声明:

① 本站未注明“稿件来源”的信息均来自网络整理。其文字、图片和音视频稿件的所属权归原作者所有。本站收集整理出于非商业性的教育和科研之目的,并不意味着本站赞同其观点或证实其内容的真实性。仅作为临时的测试数据,供内部测试之用。本站并未授权任何人以任何方式主动获取本站任何信息。

② 本站未注明“稿件来源”的临时测试数据将在测试完成后最终做删除处理。有问题或投稿请发送至: 邮箱/279061341@qq.com QQ/279061341

软考中级精品资料免费领

  • 历年真题答案解析
  • 备考技巧名师总结
  • 高频考点精准押题
  • 2024年上半年信息系统项目管理师第二批次真题及答案解析(完整版)

    难度     813人已做
    查看
  • 【考后总结】2024年5月26日信息系统项目管理师第2批次考情分析

    难度     354人已做
    查看
  • 【考后总结】2024年5月25日信息系统项目管理师第1批次考情分析

    难度     318人已做
    查看
  • 2024年上半年软考高项第一、二批次真题考点汇总(完整版)

    难度     435人已做
    查看
  • 2024年上半年系统架构设计师考试综合知识真题

    难度     224人已做
    查看

相关文章

发现更多好内容

猜你喜欢

AI推送时光机
位置:首页-资讯-后端开发
咦!没有更多了?去看看其它编程学习网 内容吧
首页课程
资料下载
问答资讯