热门标签 | HotTags
当前位置:  开发笔记 > 编程语言 > 正文

【Spark2.0系列】:Catalog和自定义Optimizer

【Spark2.0系列】适合初学

Spark 2.0系列第一篇见【Spark 2.0系列】: Spark Session API和Dataset API,本文将讲解Spark 2.0 的Catalog 和Custom Optimizer。

首先,先了解下RDD 和Dataset 在开发中使用对比。

RDD 和Dataset 使用对比

Dataset API 是RDD 和DataFrame API 的统一,但大部分Dataset API 与RDD API使用方法看起来是相似的(其实实现方法是不同的)。所以RDD代码很容易转换成Dataset API。下面直接上代码:

WordCount
  • RDD

val rdd = sparkContext.textFile("src/main/resources/data.txt")

 

val wordsRDD = rdd.flatMap(value => value.split("\\s+"))

val wordsPair = wordsRDD.map(word => (word,1))

val wordCount = wordsPair.reduceByKey(_+_)

  • Dataset

val ds = sparkSession.read.text("src/main/resources/data.txt")

import sparkSession.implicits._

val wordsDs = ds.flatMap(value => value.split("\\s+"))

val wordsPairDs = wordsDs.groupByKey(value => value)

val wordCountDs = wordsPairDs.count()

其它


RDDDataset
Cachingrdd.cache()ds.cache()
Filter

val filteredRDD = wordsRDD.filter(value => value ==”hello”)

val filteredDS = wordsDs.filter(value => value ==”hello”)
Map Partition

val mapPartitiOnsRDD= rdd.mapPartitions(iterator => List(iterator.count(value => true)).iterator)

val mapPartitiOnsDs= ds.mapPartitions(iterator => List(iterator.count(value => true)).iterator)
reduceByKeyval reduceCountByRDD = wordsPair.reduceByKey(+)val reduceCountByDs = wordsPairDs.mapGroups((key,values) =>(key,values.length))
备注:此处表格横屏观看效果更佳。

Dataset 和RDD 相互转换
  • RDD

val dsToRDD = ds.rdd

  • Dataset

RDD 转换成Dataframe稍麻烦,需要指定schema。

val rddStringToRowRDD = rdd.map(value => Row(value))

val dfschema = StructType(Array(StructField("value",StringType)))

val rddToDF = sparkSession.createDataFrame(rddStringToRowRDD,dfschema)

val rDDToDataSet = rddToDF.as[String]

Catalog API

DataSet 和Dataframe API 支持结构化数据分析,而结构化数据重要的是管理metadata。这里的metadata包括temporary metadata(临时表);registered udfs;permanent metadata(Hive metadata或HCatalog)。

早期Spark版本并未提供标准的API访问metadata,开发者需要使用类似show tables的查询来查询metadata;而Spark 2.0 在Spark SQL中提供标准API 调用catalog来访问metadata。

访问Catalog

建立SparkSession,然后调用Catalog:

val catalog = sparkSession.catalog

查询数据库

catalog.listDatabases().select("name").show()

listDatabases可查询所有数据库。在Hive中,Catalog可以访问Hive metadata中的数据库。listDatabases返回一个dataset,所以你可以使用适用于dataset的所有操作去处理metadata。

用createTempView 注册Dataframe

早期版本Spark用registerTempTable注册dataframe,而Spark 2.0 用createTempView替代。

df.createTempView("sales")

一旦注册视图,即可使用listTables访问所有表。

查询表

catalog.listTables().select("name").show()

检查表缓存

通过Catalog可检查表是否缓存。访问频繁的表缓存起来是非常有用的。

catalog.isCached("sales")

默认表是不缓存的,所以你会得到false。

df.cache()

catalog.isCached("sales")

现在将会打印true。

删除视图

catalog.dropTempView("sales")

查询注册函数

catalog.listFunctions().

select("name","description","className","isTemporary").show(100)

Catalog不仅能查询表,也可以访问UDF。上面代码会显示Spark Session中所有的注册函数(包括内建函数)。

自定义 Optimizer
Catalyst optimizer

Spark SQL使用Catalyst优化所有的查询,优化之后的查询比直接操作RDD速度要快。Catalyst是基于rule的,每个rule都有一个特定optimization,比如,ConstantFolding rule用来移除常数表达式,具体可直接看Spark SQL源代码。

在早期版本Spark中,如果想自定义optimization,需要开发者修改Spark源代码。操作起来麻烦,而且要求开发者能读懂源码。在Spark 2.0中,已提供API自定义optimization。

访问Optimized plan

在开始编写自定义optimization之前,先来看看如何访问optimized plan:

val df = sparkSession.read.option("header","true").csv("src/main/resources/data.csv")

val multipliedDF = df.selectExpr("amountPaid * 1")

println(multipliedDF.queryExecution.optimizedPlan.numberedTreeString)

上面的代码是加载一个csv文件,并对某一行所有值乘以1。queryExecution 可访问查询相关的所有执行信息。 queryExecutionoptimizedPlan对象可以访问dataframe的optimized plan。

Spark中的执行计划以tree表示,所以用numberedTreeString打印optimized plan。打印结果如下:

00 Project [(cast(amountPaid#3 as double) * 1.0) AS (amountPaid * 1)#5]01 +- Relation[transactionId#0,customerId#1,itemId#2,amountPaid#3] csv

所有执行计划是由底向上读取:

  • 01 Relation - 从csv 文件建立一个dataframe

  • 00 Project - 投影操作

编写自定义optimizer rule

从上面的执行计划可以清晰的看到:对一列的每个值乘以1 这里并没有优化。我们知道,乘以1 这个操作应该返回的是值本身,所以可以利用这个特点来增加只能点的optimizer。代码如下:

object MultiplyOptimizationRule extends Rule[LogicalPlan] {

   def apply(plan: LogicalPlan): LogicalPlan = plan transformAllExpressions {

     case Multiply(left,right) if right.isInstanceOf[Literal] &&

       right.asInstanceOf[Literal].value.asInstanceOf[Double] == 1.0 =>

       println("optimization of one applied")

       left

   }

 }

这里MultiplyOptimizationRule扩展自Rule类,采用Scala的模式匹配编写。检测右操作数是否是 1,如果是1 则直接返回左节点。

把MultiplyOptimizationRule加入进optimizer:

sparkSession.experimental.extraOptimizatiOns= Seq(MultiplyOptimizationRule)

你可以使用extraOptimizations将定义好的Rule加入 catalyst。

下面实际使用看看效果:

val multipliedDFWithOptimization = df.selectExpr("amountPaid * 1")

println("after optimization")

println(multipliedDFWithOptimization.queryExecution.

optimizedPlan.numberedTreeString)

我们看到打印结果:

00 Project [cast(amountPaid#3 as double) AS (amountPaid * 1)#7]01 +- Relation[transactionId#0,customerId#1,itemId#2,amountPaid#3] csv

说明自定义Optimizer已生效。


侠天,专注于大数据、机器学习和数学相关的内容,并有个人公众号:bigdata_ny分享相关技术文章。

若发现以上文章有任何不妥,请联系我。





推荐阅读
author-avatar
xeyuxing369
这个家伙很懒,什么也没留下!
PHP1.CN | 中国最专业的PHP中文社区 | DevBox开发工具箱 | json解析格式化 |PHP资讯 | PHP教程 | 数据库技术 | 服务器技术 | 前端开发技术 | PHP框架 | 开发工具 | 在线工具
Copyright © 1998 - 2020 PHP1.CN. All Rights Reserved | 京公网安备 11010802041100号 | 京ICP备19059560号-4 | PHP1.CN 第一PHP社区 版权所有