周天亮

周天亮

关注
发信
关注(28)粉丝(399)

40-天亮大数据系列教程之Spark常用算子分析与应用

目录
1、算子概述
2、Spark算子介绍与应用
3、经典算子练习
详情
1、算子概述

  • 什么是算子

    • 英文翻译为：Operator（简称op）

    • 狭义：指从一个函数空间到另一个函数空间（或它自身）的映射。

    • 广义：指从一个空间到另一个空间的映射

    • 通俗理解：指事物(数据或函数)从一个状态到另外一个状态的过程抽象。
      实质就是映射，就是关系，就是变换。* 算子的重要作用

    • 算子越少，灵活性越低，则实现相同功能的编程复杂度越高，算子越多则反之。

      • 老手机与智能手机

        • 老电脑与新电脑之HDMI (跟VGA是对等的)

      • 算子越少，表现力越差，面对复杂场景则易用性较差。算子越多的则反之。

        • 黑白彩电与彩色电视
        • 彩色电视和智能电视

    • MapReduce 与 Spark算子比较

      • MapReduce只有2个算子，Map和Reduce，绝大多数应用场景下，均需要复杂编码才能达到用户需求。
      • Spark有80多个算子，进行充分的组合应用后，能满足绝大多数的应用场景。

2、Spark算子介绍与应用

  • 2.1 算子分类

    • 1）转换算子（Transformation）：此种算子不触发提交作业，只有作业被提交后才会真正启动转换计算。

    • Value型转换算子 ： 其处理的数据项是Value型

    • 输入分区与输出分区一对一型

      • map算子
      • flatMap算子
      • mapPartitions算子
      • glom算子
  • 输入分区与输出分区多对一型

    • union算子

    • cartesian算子

    • 输入分区与输出分区多对多型
      * grouBy算子
      * 输出分区为输入分区子集型

        	* filter算子
  	* distinct算子
  	* subtract算子
  	* sample算子
  	* takeSample算子
  * Cache型

  	* cache算子
  	* persist算子

      • Key-Value型转换算子：其处理的数据是Key-Value型

        • 输入分区与输出分区一对一

          • mapValues算子

        • 对单个RDD聚集

          • combineByKey算子
          • reduceByKey算子
          • partitionBy算子

        • 对两个RDD聚集

          • cogroup算子

        • 连接

          • join算子
          • leftOutJoin算子
          • rightOutJoin算子

    • 2） 行动算子(Action）：此种算子会触发SparkContext提交作业。

      • 无输出(是指不输出hdfs、本地文件当中)

        • foreach算子

      • HDFS

        • saveAsTextFile算子
        • saveAsObjectFile算子

      • Scala集合和数据类型

        • collect算子
        • collectAsMap算子
        • reduceByKeyLocally算子
        • lookup算子
        • count算子
        • top算子
        • reduce算子
        • fold算子
        • aggregate算子

    • 2.2 常用算子分析与应用

      • 2.2.1 Value型转换算子

          1. map

            • 类比于mapreduce中的map操作，给定一个输入通过map函数映到成一个新的元素输出

    • case_1
      val first = sc.parallelize(List(“Hello”,“World”,“天亮教育”,“大数据”),2)val second= first.map(_.length)second.collect

    • case_2
      val first = sc.parallelize(1 to 5,2)first.map(1 to _).collect

      1. flatMap

      • 给定一个输入，将返回的所有结果打平成一个一维集合结构

        • case_1

val first = sc.parallelize(1 to 5,2)first.flatMap(1 to _).collect
* case_2

val first = sc.parallelize(List(“one”,“two”,“three”),2)first.flatMap(x => List(x,x,x)).collect
* case_3

val first = sc.parallelize(List(“one”,“two”,“three”),2)first.flatMap(x => List(x+"_1",x+"_2",x+"_3")).collect
* 3) mapPartitions

	* 以分区为单位进行计算处理，而map是以每个元素为单位进行计算处理。
	* 当在map过程中需要频繁创建额外对象时，如文件输出流操作、jdbc操作、Socket操作等时，当用mapPartitions算子

		* case_1

val rdd=sc.parallelize(Seq(1,2,3,4,5),3)var rdd2=rdd.mapPartitions(partition=>{ partition.map(num => num * num) })rdd2.max
* case_2

val rdd=sc.parallelize(Seq(1,2,3,4,5),3)var rdd2=rdd.mapPartitions(partition=>{ partition.flatMap(1 to _) })rdd2.count
* 4) glom

	* 以分区为单位，将每个分区的值形成一个数组

val a = sc.parallelize(Seq(“one”,“two”,“three”,“four”,“five”,“six”,“seven”),3)a.glom.collect
* 5) union算子

	* 将两个RDD合并成一个RDD，并不去重
	* 会发生多分区合并成一个分区的情况

val a = sc.parallelize(1 to 4, 2)val b = sc.parallelize(3 to 6, 2)a.union(b).collect(a ++ b).collect(a union b).collect
* 6) groupBy算子

	* 输入分区与输出分区多对多型

val a = sc.parallelize(Seq(1,3,4,5,9,100,200), 3)a.groupBy(x => { if (x > 10) “>10” else “<=10” }).collect
* 7) filter算子

	* 输出分区为输入分区子集型

val a = sc.parallelize(1 to 21, 3)val b = a.filter(_ % 4 == 0)b.collect
* 8 ) distinct算子

	* 输出分区为输入分区子集型

		* case_1

val a = sc.parallelize(1 to 4, 2)val b = sc.parallelize(3 to 6, 2)a.union(b).distinct().collect
* case_2

val c = sc.parallelize(List(“张三”, “李四”, “李四”, “王五”), 2)c.distinct.collect
* 9) cache算子

	* cache 将 RDD 元素从磁盘缓存到内存。 相当于 persist(MEMORY_ONLY) 函数的功能。
	* 主要应用在当RDD数据反复被使用的场景下

		* case_1

val a = sc.parallelize(1 to 4, 2)val b = sc.parallelize(3 to 6, 2)a.union(b).counta.union(b).distinct().collect
* case_2

val a = sc.parallelize(1 to 4, 2)val b = sc.parallelize(3 to 6, 2)val c=a.union(b).cachec.countc.distinct().collect
* 2.2.2 Key-Value型转换算子：其处理的数据是Key-Value型

	* mapValues算子

		* 输入分区与输出分区一对一
		* 针对（Key， Value）型数据中的 Value 进行 Map 操作，而不对 Key 进行处理。

val first = sc.parallelize(List((“张一”,1),(“张二”,2),(“张三”,3),(“张四”,4)),2)val second= first.mapValues(x=>x+1)second.collect
* combineByKey算子

	* 定义

def combineByKey[C](createCombiner: (V) => C,mergeValue: (C, V) => C,mergeCombiners: (C, C) => C): RDD[(String, C)]
* createCombiner: 对每个分区内的同组内元素如何聚合，形成一个累加器
* mergeValue: 将前边的累加器与新遇到的值进行合并的方法
* mergeCombiners: 每个分区都是独立处理，故同一个键可以有多个累加器。如果有两个或者更

  多的分区都有对应同一个键的累加器，用方法将各个分区的结果进行合并。
* case_1

val first = sc.parallelize(List((“张一”,1),(“李一”,1),(“张一”,2),(“张一”,3),(“李一”,3),(“李三”,3),(“张四”,4)),2)val second= first.combineByKey(List(_), (x:List[Int], y:Int) => y :: x, (x:List[Int], y:List[Int]) => x ::: y)second.collect
* reduceByKey算子

	* 按key聚合后对组进行归约处理，如求和、连接等操作

val first = sc.parallelize(List(“小米”, “华为”, “小米”, “小米”, “华为”, “苹果”), 2)val second = first.map(x => (x, 1))second.reduceByKey(_ + _).collect
* join算子

	* 对K-V结构的RDD进行按K的join操作，最后将V部分做flat打平操作。

val first = sc.parallelize(List((“张一”,11),(“李二”,12)),2)val second = sc.parallelize(List((“张一”,21),(“李二”,22),(“王五”,23)),2)first.join(second).collect
* 2.2.3 行动算子(Action）：此种算子会触发SparkContext提交作业。触发了RDD DAG 的执行。
* 无输出型：不落地到文件或是hdfs的作用

	* foreach算子

val first = sc.parallelize(List(“小米”, “华为”, “小米”, “小米”, “华为”, “苹果”), 2)first.foreach(println _)
* HDFS输出型

	* saveAsTextFile算子

val first = sc.parallelize(List(“小米”, “华为”, “小米”, “小米”, “华为”, “苹果”), 2)//指定本地保存的目录first.saveAsTextFile(“file:///home/spark/text”) //指定hdfs保存的目录，默认亦保存在hdfs中first.saveAsTextFile(“spark_shell_output_1”)
* Scala集合和数据类型

	* collect算子

		* 相当于toArray操作，将分布式RDD返回成为一个scala array数组结果，实际是Driver所在的机器节点，再针对该结果操作

val first = sc.parallelize(List(“小米”, “华为”, “小米”, “小米”, “华为”, “苹果”), 2)first.collect
* collectAsMap算子

	* 相当于toMap操作，将分布式RDD的kv对形式返回成为一个的scala map集合，实际是Driver所在的机器节点，再针对该结果操作

val first = sc.parallelize(List((“张一”,1),(“李一”,1),(“张一”,2),(“张一”,3),(“李一”,3),(“李三”,3),(“张四”,4)),2)first.collectAsMap
* lookup算子

	* 对（Key，Value）型的RDD操作，返回指定Key对应的元素形成的Seq。

val first = sc.parallelize(List(“小米”, “华为”, “华米”, “大米”, “苹果”,“米老鼠”), 2)val second=first.map(x=>({if(x.contains(“米”)) “有米” else “无米”},x))second.lookup(“有米”)
* reduce算子

	* 先对两个元素进行reduce函数操作，然后将结果和迭代器取出的下一个元素进行reduce函数操作，直到迭代器遍历完所有元素，得到最后结果。

//求value型列表的和val a = sc.parallelize(1 to 10, 2)a.reduce(_ + _)//求key-value型列表的value的和val a = sc.parallelize(List((“one”,1),(“two”,2),(“three”,3),(“four”,4)), 2)a.reduce((x,y)=>(“sum”,x._2 + y._2))._2
* fold算子

	* fold算子签名：  def fold(zeroValue: T)(op: (T, T) => T): T
	* 其实就是先对rdd分区的每一个分区进行op函数，在调用op函数过程中将zeroValue参与计算，最后在对所有分区的结果调用op函数，同理此处zeroValue再次参与计算。

//和是41,公式=(1+2+3+4+5+6+10)+10sc.parallelize(List(1, 2, 3, 4, 5, 6), 1).fold(10)(+)//和是51,公式=(1+2+3+10)+(4+5+6+10)+10=51sc.parallelize(List(1, 2, 3, 4, 5, 6), 2).fold(10)(+)//和是61,公式=(1+2+10)+(3+4+10)+(5+6+10)+10=61sc.parallelize(List(1, 2, 3, 4, 5, 6), 3).fold(10)(+)3、经典算子练习
* cartesian算子
* subtract算子
* sample算子
* takeSample算子
* persist算子
* cogroup算子
* leftOutJoin算子
* rightOutJoin算子
* saveAsObjectFile算子
* count算子
* top算子
* aggregate算子

天亮教育是一家从事大数据云计算、人工智能、教育培训、产品开发、咨询服务、人才优选为一体的综合型互联网科技公司。
公司由一批BAT等一线互联网IT精英人士创建，
以"快乐工作，认真生活，打造高端职业技能教育的一面旗帜"为愿景，胸怀"让天下没有难找的工作"使命，
坚持"客户第一、诚信、激情、拥抱变化"的价值观，
全心全意为学员赋能提效，践行技术改变命运的初心。

更多学习讨论， 请加入
官方-天亮大数据交流-366784928
群二维码：
40-天亮大数据系列教程之Spark常用算子分析与应用-LMLPHP
天亮教育公开课-从小白到大佬修成记-全系列视频地址：http://bbs.myhope365.com/forum.php?mod=viewthread&tid=1422&extra=page%3D1

欢迎关注天亮教育公众号，大数据技术资料与课程、招生就业动态、教育资讯动态、创业历程分享一站式分享，官方微信公众号二维码：
40-天亮大数据系列教程之Spark常用算子分析与应用-LMLPHP

天亮教育官方群318971238，
爬虫、nlp技术qq群320349384
hadoop & spark & hive技术群297585251，
教育培训官网：http://myhope365.com
项目研发官网：http://shangyuninfo.com/
官方天亮论坛:http://bbs.myhope365.com/

12-28 19:40
Powered by LMLPHP ©2024 周天亮 0.045613