spark初识

一.简介

1.什么是spark

Apache Spark 是专为大规模数据处理而设计的快速通用的计算引擎。Spark是UC Berkeley AMP lab (加州大学伯克利分校的AMP实验室)所开源的类Hadoop MapReduce的通用并行计算框架,Spark拥有Hadoop MapReduce所具有的优点;但不同于MapReduce的是Job中间输出结果可以保存在内存中,从而不再需要读写HDFS,因此Spark能更好地适用于数据挖掘与机器学习等需要迭代的MapReduce的算法。

Spark是Scala编写,方便快速编程。

2.总体栈技术讲解

3.spark与MapReduce的区别

都是分布式计算框架,Spark基于内存,MR基于HDFS。Spark处理数据的能力一般是MR的十倍以上,Spark中除了基于内存计算外,还有DAG有向无环图来切分任务的执行先后顺序。

5.spark运行模式

  • local

    • 多用于本地测试

  • standalone

    • spark自带的一个资源调度框架,支持完全分布式

  • yarn

    • spark也可以基于yarn来计算

  • Mesos

    • 资源调度框架

  •  要基于Yarn来进行资源调度,必须实现AppalicationMaster接口,Spark实现了这个接口,所以可以基于Yarn。

二.sparkcore

1.RDD

  • RDD(Resilient Distributed Dataset),弹性分布式数据集

  • 五大特性

    • RDD是由一系列的partition组成的

    • 函数是作用在每一个partition上的

    • RDD之间是有一系列的依赖关系

    • 分区器是作用在K,V格式的RDD上

    • RDD提供一系列最佳的计算位置

  • RDD理解图

  • 注意

    • textFile方法底层封装的是MR读取文件的方式,读取文件之前先split,默认split大小是一个块的大小

  • 什么是K,V格式的RDD

    • 如果RDD里面存储的数据都是二元组对象,那么这个RDD我们就叫做K,V格式的RDD

  • 哪里体现RDD的弹性(容错性)

    • partition数量和大小没有限制,体现了RDD的弹性

    • RDD之间依赖关系,可以基于上一个RDD重新计算出RDD

  • 哪里体现RDD的分布式

    • RDD是由partition组成,partition是分布在不同节点上的

  • RDD提供计算最佳位置,体现了数据本地化。体现了大数据中“计算移动数据不移动”的理念。

2.Spark任务执行原理

    • driver和worker是启动在节点上的进程,运行在JVM中的进程

    • driver与集群节点之间有频繁的通信

    • driver负责任务的分发和结果的回收,任务的调度,如果任务的计算结果非常大就不要回收了,会造成内存溢出问题

    • worker是standalone资源调度框架里面资源管理的从节点.也是JVM进程

    • Master是Standalone资源调度框架里面资源管理的主节点。也是JVM进程。

3.spark代码流程

  • 创建sparkconf对象

    • 设置application name

    • 设置运行模式及资源需求

  • 创建sparkcontext对象

  • 基于spark的上下文创建一个RDD,对RDD进行处理

  • 应用程序中要有action类算子来触发transformation类算子进行执行

  • 关闭spark上下文对象sparkcontext

  • 源码

    • object WordCount {?  def main(args: Array[String]): Unit = {    val conf = new SparkConf();    conf.setMaster("local").setAppName("wc")    val sc = new SparkContext(conf)    val lineRDD=sc.textFile("./word.txt")    val wordRDD = lineRDD.flatMap(x=>x.split(" "))    val kvRDD=wordRDD.map(x=>(x,1))    val result = kvRDD.filter(x=>{      if(x.equals("hello"))        true      else        false    })    result.foreach(println)  }}

4.Transformations转换算子

  • 转换算子输入输出类型

  • Transformations类算子是一类算子(函数)叫做转换算子,如map,flatMap,reduceByKey等。Transformations算子是延迟执行,也叫懒加载执行。

  • 转换算子介绍

    • filter

      • 过滤符合条件的记录数,true保留,false过滤掉。

    • map

      • 将一个RDD中的每个数据项,通过map中的函数映射变为一个新的元素。特点:输入一条,输出一条数据。

    • flatmap

      • 先map后flat。与map类似,每个输入项可以映射为0到多个输出项。

    • sample

      • 随机抽样算子,根据传进去的小数按比例进行又放回或者无放回的抽样。

    • reduceByKey

      • 将相同的Key根据相应的逻辑进行处理。

    • sortByKey/sortBy

      • 作用在K,V格式的RDD上,对key进行升序或者降序排序。

    • join

    • leftOuterJoin

    • rightOuterJoin

    • fullOuterJoin

    • 作用在K,V格式的RDD上,根据K进行连接,对(K,V),join(K,W)返回(K,(V,W))

    • union

      • 合并两个数据集.两个数据集的类型要一致

    • intersection

      • 取两个数据集的交集

    • subtract

      • 与map类似,遍历的单位是每个partition上的数据。

    • distinct相当于(map+reduceByKey+map)

      • 去重

    • cogroup

      • 当调用类型(K,V)和(K,W)的数据上时,返回一个数据集(K,(Iterable<V>,Iterable<W>))

    • mappartition

      • 与map类似,遍历的单位是每个partition上的数据。

    • mapPartitionWithIndex

      • 类似于mapPartitions,初次之外还会携带分区的索引值

    • repartition(宽依赖)

      • 增加或减少分区.会产生shuffle

    • coalesce(可以做宽依赖也可以做窄依赖)

      • coalesce常用来减少分区,第二个参数是减少分区的过程中是否产生shuffle。

      • true为产生shuffle,false不产生shuffle。默认是false。

        如果coalesce设置的分区数比原来的RDD的分区数还多的话,第二个参数设置为false不会起作用,如果设置成true,效果和repartition一样。

        即repartition(numPartitions) = coalesce(numPartitions,true)

    • groupByKey

      • 作用在K,V格式的RDD上。根据Key进行分组。作用在(K,V),返回(K,Iterable <V>)。

    • zip

      • 将两个RDD中的元素(KV格式/非KV格式)变成一个KV格式的RDD,两个RDD的个数必须相同

    • zipWithIndex

      • 该函数将RDD中的元素和这个元素在RDD中的索引号(从0开始)组合成(K,V)对。

5.action触发算子

  • Action类算子也是一类算子(函数)叫做行动算子,如foreach,collect,count等。Transformations类算子是延迟执行,Action类算子是触发执行。一个application应用程序中有几个Action类算子执行,就有几个job运行。

  • 算子介绍

    • count

      • 返回数据集中的元素数。会在结果计算完成后回收到Driver端。

    • take

      • 返回一个包含数据集前n个元素的集合。

    • first

      • first=take(1),返回数据集中的第一个元素。

    • foreach

      • 循环遍历数据集中的每个元素,运行相应的逻辑。

    • collect

      • 将计算结果回收到Driver端。

    • foreachpartition

      • 遍历的数据是每个partition的数据。

    • countByKey

      • 作用到K,V格式的RDD上,根据Key计数相同Key的数据集元素。

    • countByValue

      • 根据数据集每个元素相同的内容来计数。返回相同内容的元素对应的条数。

    • reduce

      • 根据聚合逻辑聚合数据集中的每个元素。

6.控制算子

  • 控制算子有三种,cache,persist,checkpoint,以上算子都可以将RDD持久化,持久化的单位是partition。cache和persist都是懒执行的。必须有一个action类算子触发执行。checkpoint算子不仅能将RDD持久化到磁盘,还能切断RDD之间的依赖关系。

  • cache

    • 默认将RDD的数据持久化到内存中。cache是懒执行

    • SparkConf conf = new SparkConf(); conf.setMaster("local").setAppName("CacheTest"); JavaSparkContext jsc = new JavaSparkContext(conf); JavaRDD<String> lines = jsc.textFile("./NASA_access_log_Aug95");? lines = lines.cache(); long startTime = System.currentTimeMillis(); long count = lines.count(); long endTime = System.currentTimeMillis(); System.out.println("共"+count+ "条数据,"+"初始化时间+cache时间+计算时间="+           (endTime-startTime));         long countStartTime = System.currentTimeMillis(); long countrResult = lines.count(); long countEndTime = System.currentTimeMillis(); System.out.println("共"+countrResult+ "条数据,"+"计算时间="+ (countEndTime-           countStartTime));         jsc.stop();
  • persist

    • 可以指定持久化的级别。最常用的是MEMORY_ONLY和MEMORY_AND_DISK。”_2”表示有副本数。

  • 持久化级别

  • cache和persist的注意事项

    • cache和persist都是懒执行,必须有一个action类算子触发执行。

    • cache和persist算子的返回值可以赋值给一个变量,在其他job中直接使用这个变量就是使用持久化的数据了。持久化的单位是partition

    • ache和persist算子后不能立即紧跟action算子。

  • checkpoint

    • checkpoint将RDD持久化到磁盘,还可以切断RDD之间的依赖关系。

    • 执行原理

      • 当RDD的job执行完毕后,会从finalRDD从后往前回溯。

      • 当回溯到某一个RDD调用了checkpoint方法,会对当前的RDD做一个标记。

      • Spark框架会自动启动一个新的job,重新计算这个RDD的数据,将数据持久化到HDFS上。

  • 优化

    • SparkConf conf = new SparkConf(); conf.setMaster("local").setAppName("checkpoint"); JavaSparkContext sc = new JavaSparkContext(conf); sc.setCheckpointDir("./checkpoint"); JavaRDD<Integer> parallelize = sc.parallelize(Arrays.asList(1,2,3)); parallelize.checkpoint(); parallelize.count(); sc.stop();

三.集群搭建

  • 上传压缩包并解压

  • 进入安装包的conf目录下,修改slaves.template文件,先把slaves.template文件名改成 slaves

    • 然后添加从节点。保存。

    • 修改spark-env.sh.template 文件名为 spark-env.sh

    • 然后编辑spark-env.sh文件,进行配置:

      • 相关参数介绍

        • JAVA_HOME:配置java_home路径SPARK_MASTER_HOST:master的ipSPARK_MASTER_PORT:提交任务的端口,默认是7077SPARK_WORKER_CORES:每个worker从节点能够支配的core的个数SPARK_WORKER_MEMORY:每个worker从节点能够支配的内存数
  • 同步到其他节点上

  • 启动

    • 进入sbin目录下,执行当前目录下的./start-all.sh

  • 测试

    • 访问master:8080端口: