Spark(二)算子详解

目录

  • Spark(二)算子讲解

    • 一、wordcountcount
    • 二、编程模型
    • 三、RDD数据集和算子的使用

Spark(二)算子讲解

@

一、wordcountcount

基于上次的wordcount,我们来写一个wordcountcount,来对wc程序进行第二次计数,我们来分析一下性能。

package com.littlepage.wc

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

object WordCount {
  def main(args: Array[String]): Unit = {
    val conf=new SparkConf().setAppName("wc").setMaster("local")
    val sparkContext=new SparkContext(conf)
    sparkContext.setLogLevel("error")
    val fileRDD:RDD[String] = sparkContext.textFile("data/data")
    val words:RDD[String] = fileRDD.flatMap(_.split(" "))
    val pairWord:RDD[(String,Int)] = words.map((_,1))
    val res:RDD[(String,Int)] = pairWord.reduceByKey(_+_)
    println("wordcount:")
    res.foreach(println)
    val rev:RDD[(Int,Int)] = res.map((x)=>{(x._2,1)})
    val pl:RDD[(Int,Int)] = rev.reduceByKey(_+_)
    println("\nwordcountcount")
    pl.foreach(println)
    Thread.sleep(100000000)
  }
}

通过性能图,我们可以知道:

1.Spark如果不对其结果进行存储或输出,那么Spark将不会处理map或者reduce操作

2.如果进行重复输出,共用的map或者reduce操作只执行一次

3.默认如果产生一次shuffle是去查看图表的一次拐弯,为了尽量减少性能的消耗,编写程序时应该尽量减少shuffle的次数

二、编程模型

Spark编程模型和MapReduce相比,Spark可以多个Job,多个State进行执行。

源码部分参考视频

三、RDD数据集和算子的使用

1.三个必备算子

我们在写一个Spark程序中,不可避免的算子有三个,创建算子,转换算子,收集算子。

创建算子可以创建一个RDD数据集,这个创建可以在内存中(集合容器),也可以在硬盘中(文件)获取

转换算子可以处理一个RDD数据集,即map和reduce操作,都算做转换算子。

收集算子我们在写一个RDD数据集的时候,必须使用收集算子进行收集,否则不会触发shuffle。

示例,三个算子写一个过滤数字程序。

package com.littlepage

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

object demo2 {
  def main(args: Array[String]): Unit = {
    val conf=new SparkConf().setAppName("demo2").setMaster("local")
    val sc=new SparkContext(conf)
    sc.setLogLevel("error")
    val dataRDD: RDD[Int] = sc.parallelize(List(1,2,3,4,5,6,7,6,5,4,3,2,1))//创建算子
    val filterRDD: RDD[Int] = dataRDD.filter(_>3)//转换算子
    val ints:Array[Int] = filterRDD.collect()//收集算子
    Thread.sleep(100000)
  }
}
package com.littlepage

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

object demo2 {
  def main(args: Array[String]): Unit = {
    val conf=new SparkConf().setAppName("demo2").setMaster("local")
    val sc=new SparkContext(conf)
    sc.setLogLevel("error")
    val dataRDD: RDD[Int] = sc.parallelize(List(1,2,3,4,5,6,7,6,5,4,3,2,1))//创建算子
    val filterRDD: RDD[Int] = dataRDD.filter(_>3)//转换算子
    val ints:Array[Int] = filterRDD.collect()//收集算子
    Thread.sleep(100000)
  }
}
2.常见算子(交并差笛卡尔,cogroup,join)

2.1.union算子

将两个数据集合并为一个数据集,直接合并,不会产生shuffle

object union {
  def main(args: Array[String]): Unit = {
    val sc=new SparkContext(new SparkConf().setMaster("local").setAppName("union"))
    sc.setLogLevel("error")
    val rdd1:RDD[Int] = sc.parallelize(List(1,2,3,4,6,7))
    val rdd2:RDD[Int] = sc.parallelize(List(2,3,4,5))
    val uniondata = rdd1.union(rdd2)
    uniondata.foreach(print)
    Thread.sleep(100000)
  }
}

2.2.intersection算子

将2个数据集取交集,产生一个shuffle

val interdata:RDD[Int] = rdd1.intersection(rdd2)

2.3.substract算子

将2个数据集取差集,产生一个shuffle

val subdata:RDD[Int] = rdd1.substract(rdd2)

2.4.cartesian算子

将2个数据集取笛卡尔积,不产生shuffle

val cartesiandata:RDD[Int] = rdd1.cartesian(rdd2)

2.5.cogroup算子

两个分组进行,key作为结果的key,value集合进行一个二元祖,包含两个分区的元素,产生一个shuffle。

val rdd1:RDD[(String,Int)] = sc.parallelize(List(
      ("zhangsan",11),
      ("zhangsan",12),
      ("lisi",13),
      ("wangwu",14)
    ));
    val rdd2:RDD[(String,Int)] = sc.parallelize(List(
      ("zhangsan",21),
      ("zhangsan",22),
      ("lisi",23),
      ("zhaoliu",28)
    ))
    val cogroupdata:RDD[(String,(Iterable[Int],Iterable[Int]))] = rdd1.cogroup(rdd2)

6.join,leftOuterJoin,rightOuterJoin,fullOuterJoin算子

val joindata:RDD[(String,(Int,Int))] = rdd1.join(rdd2)
val leftdata:RDD[(String,(Int,Option[Int]))] = rdd1.leftOuterJoin(rdd2)
val rightdata:RDD[(String,(Option[Int],Int))]  = rdd2.rightOuterJoin(rdd2)
val fulldata:RDD[(String,(Option[Int],Option[Int]))]  = rdd1.fullOuterJoin(rdd2)
3.排序和聚合计算

3.1.swap算子

将一个k-v数据集的key和value交换,用法

data.map(_.swap)

3.2.sort算子

sort算子可以将按照key进行全排序

data.sortByKey()

3.3.take算子

获得数据的前n个,n为一个整型

data.take(n)

3.4.distinct去重

去除key相同的

val keys:RDD[(String,String) = map.distinct()

原文地址:https://www.cnblogs.com/littlepage/p/11751867.html

时间: 2024-10-10 10:27:19

Spark(二)算子详解的相关文章

iOS 开发之照片框架详解之二 —— PhotoKit 详解(下)

这里接着前文<iOS 开发之照片框架详解之二 —— PhotoKit 详解(上)>,主要是干货环节,列举了如何基于 PhotoKit 与 AlAssetLibrary 封装出通用的方法. 三. 常用方法的封装 虽然 PhotoKit 的功能强大很多,但基于兼容 iOS 8.0 以下版本的考虑,暂时可能仍无法抛弃 ALAssetLibrary,这时候一个比较好的方案是基于 ALAssetLibrary 和 PhotoKit 封装出一系列模拟系统 Asset 类的自定义类,然后在其中封装好兼容 A

iOS 开发之照片框架详解之二 —— PhotoKit 详解(上)

一. 概况 本文接着 iOS 开发之照片框架详解,侧重介绍在前文中简单介绍过的 PhotoKit 及其与 ALAssetLibrary 的差异,以及如何基于 PhotoKit 与 AlAssetLibrary 封装出通用的方法. 这里引用一下前文中对 PhotoKit 基本构成的介绍: PHAsset: 代表照片库中的一个资源,跟 ALAsset 类似,通过 PHAsset 可以获取和保存资源 PHFetchOptions: 获取资源时的参数,可以传 nil,即使用系统默认值 PHAssetCo

一步一步造个Ioc轮子(二),详解泛型工厂

一步一步造个Ioc轮子目录 .net core发布了,一步一步造个Ioc轮子,弄点.net魔法,近new的速度(一) 一步一步造个Ioc轮子(二),详解泛型工厂 详解泛型工厂 既然我说Ioc容器就是一个豪华版工厂,自动化装配的工厂,那我们就从工厂入手吧,先造个工厂,然后升级成Ioc容器 首先我们来写一个最最最简单的抽象工厂类,还是以前一篇的短信为例 public class SMSFactory { public static ISMS Get() { return new XSMS(); }

Apache Spark源码走读之16 -- spark repl实现详解

欢迎转载,转载请注明出处,徽沪一郎. 概要 之所以对spark shell的内部实现产生兴趣全部缘于好奇代码的编译加载过程,scala是需要编译才能执行的语言,但提供的scala repl可以实现代码的实时交互式执行,这是为什么呢? 既然scala已经提供了repl,为什么spark还要自己单独搞一套spark repl,这其中的缘由到底何在? 显然,这些都是问题,要解开这些谜团,只有再次开启一段源码分析之旅了. 全局视图 上图显示了java源文件从编译到加载执行的全局视图,整个过程中最主要的步

ViewPager 详解(二)---详解四大函数

前言:上篇中我们讲解了如何快速实现了一个滑动页面,但问题在于,PageAdapter必须要重写的四个函数,它们都各有什么意义,在上节的函数内部为什么要这么实现,下面我们就结合Android的API说明,详细讲解一下. 相关文章: 1.<ViewPager 详解(一)---基本入门> 2.<ViewPager 详解(二)---详解四大函数> 3.<ViewPager 详解(三)---PagerTabStrip与PagerTitleStrip添加标题栏的异同> 4.<

Apache Spark 内存管理详解

Spark 作为一个基于内存的分布式计算引擎,其内存管理模块在整个系统中扮演着非常重要的角色.理解 Spark 内存管理的基本原理,有助于更好地开发 Spark 应用程序和进行性能调优.本文旨在梳理出 Spark 内存管理的脉络,抛砖引玉,引出读者对这个话题的深入探讨.本文中阐述的原理基于 Spark 2.1 版本,阅读本文需要读者有一定的 Spark 和 Java 基础,了解 RDD.Shuffle.JVM 等相关概念. 在执行 Spark 的应用程序时,Spark 集群会启动 Driver

Struts2学习笔记二 配置详解

Struts2执行流程 1.简单执行流程,如下所示: 在浏览器输入请求地址,首先会被过滤器处理,然后查找主配置文件,然后根据地址栏中输入的/hello去每个package中查找为/hello的namespace,然后在包下寻找名为HelloAction的action,反射创建控制器HelloAction对象,调用method指定的方法hello,拿到返回值“success”,去result的配置节找success对应的页面(hello.js),呈现给用户. 2.Struts的架构,具体参考这里

Spring二 Bean详解

Bean详解 Spring框架的本质其实是:通过XML配置来驱动Java代码,这样就可以把原本由java代码管理的耦合关系,提取到XML配置文件中管理.这样就实现了系统中各组件的解耦,有利于后期的升级和维护.1.Bean的基本定义和Bean别名<beans>元素是Spring配置文件的根元素,该元素可以指定如下属性:default-lazy-init:指定<beans>元素下配置的所有bean默认的延迟初始化行为default-merge:指定<beans>元素下配置的

Struts2学习笔记(二)——配置详解

1.Struts2配置文件加载顺序: default.properties(默认常量配置) struts-default.xml(默认配置文件,主要配置bean和拦截器) struts-plugin.xml(配置插件) struts.xml(配置action或者常量等) struts.properties(常量配置) web.xml(配置JavaEE,如:监听器和过滤器) 2.Struts2配置文件详解 1)default.properties default.properties是Struts

Hibernate入门(二)——hibernateAPI详解

Hibernate API 详解 1.Configuration 功能:配置加载类,用于加载主配置,orm元数据加载 .创建: Configuration conf = new Configuration(); 读取指定配置文件(加载主配置文件,即我们经常创建的"hibernate.cfg.xml") 从下图中可以发现有很多关于读取方法的重载... 虽然有这么多重载,但是一般咱就用无参构造方法把,默认找到src下的hibernate.cfg.xml文件 conf.configure()