目录
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