专栏首页米虫的家BigData--大数据分析引擎Spark

BigData--大数据分析引擎Spark

一、Spark运行

1、Spark内置模块

  • Spark Core:实现了Spark的基本功能,包含任务调度、内存管理、错误恢复、与存储系统交互等模块。Spark Core中还包含了对弹性分布式数据集(Resilient Distributed DataSet,简称RDD)的API定义。
  • Spark SQL:是Spark用来操作结构化数据的程序包。通过Spark SQL,我们可以使用 SQL或者Apache Hive版本的SQL方言(HQL)来查询数据。Spark SQL支持多种数据源,比如Hive表、Parquet以及JSON等。
  • Spark Streaming:是Spark提供的对实时数据进行流式计算的组件。提供了用来操作数据流的API,并且与Spark Core中的 RDD API高度对应。
  • Spark MLlib:提供常见的机器学习(ML)功能的程序库。包括分类、回归、聚类、协同过滤等,还提供了模型评估、数据 导入等额外的支持功能。
  • 集群管理器:Spark 设计为可以高效地在一个计算节点到数千个计算节点之间伸缩计 算。为了实现这样的要求,同时获得最大灵活性,Spark支持在各种集群管理器(Cluster Manager)上运行,包括Hadoop YARN、Apache Mesos,以及Spark自带的一个简易调度 器,叫作独立调度器

2、spark通用运行建议流程

  • textFile(“input”):读取本地文件input文件夹数据;
  • flatMap(_.split(“ “)):压平操作,按照空格分割符将一行数据映射成一个个单词;
  • map((_,1)):对每一个元素操作,将单词映射为元组;
  • reduceByKey(+):按照key将值进行聚合,相加;
  • collect:将数据收集到Driver端展示。

3、Spark和yarn联动

二、RDD转换

1、 Value类型

1)map(func)

返回一个新的RDD,该RDD由每一个输入元素经过func函数转换后组成

2)mapPartitions(func)

类似于map,但独立地在RDD的每一个分片上运行,因此在类型为T的RDD上运行时,func的函数类型必须是Iterator[T] => Iterator[U]。假设有N个元素,有M个分区,那么map的函数的将被调用N次,而mapPartitions被调用M次,一个函数一次处理所有分区。

3)mapPartitionsWithIndex(func)

类似于mapPartitions,但func带有一个整数参数表示分片的索引值,因此在类型为T的RDD上运行时,func的函数类型必须是(Int, Interator[T]) => Iterator[U];

map()和mapPartition()的区别

  • map():每次处理一条数据。
  • mapPartition():每次处理一个分区的数据,这个分区的数据处理完后,原RDD中分区的数据才能释放,可能导致OOM。

4)flatMap(func)

类似于map,但是每一个输入元素可以被映射为0或多个输出元素(所以func应该返回一个序列,而不是单一元素)

scala

val config = new SparkConf().setMaster("local[*]").setAppName("WordCount")

val sc = new SparkContext(config)

val listRDD: RDD[List[Int]] = sc.makeRDD(Array(List(1, 2), List(3, 4)))

val flatMapRDD = listRDD.flatMap(datas => datas)

flatMapRDD.collect().foreach(println)

5)glom

将每一个分区形成一个数组,形成新的RDD类型时RDD[Array[T]]

scala

val config = new SparkConf().setMaster("local[*]").setAppName("WordCount")

val sc = new SparkContext(config)

val listRDD = sc.makeRDD(1 to 16, 4)

val flatMapRDD = listRDD.glom()

flatMapRDD.collect().foreach(array=>{
  println(array.mkString(","))
})

6)groupBy(func)

分组,按照传入函数的返回值进行分组。将相同的key对应的值放入一个迭代器。

scala

val config = new SparkConf().setMaster("local[*]").setAppName("WordCount")

val sc = new SparkContext(config)

val listRDD = sc.makeRDD(1 to 16)

val groupByRDD = listRDD.groupBy(i => i % 2)

groupByRDD.collect().foreach(println)


//打印结果
(0,CompactBuffer(2, 4, 6, 8, 10, 12, 14, 16))
(1,CompactBuffer(1, 3, 5, 7, 9, 11, 13, 15))

7)filter(func)

过滤。返回一个新的RDD,该RDD由经过func函数计算后返回值为true的输入元素组成。

scala

val config = new SparkConf().setMaster("local[*]").setAppName("WordCount")

val sc = new SparkContext(config)

val listRDD = sc.makeRDD(1 to 16)

val filterRDD = listRDD.filter(i => (i % 2) == 0)

filterRDD.collect().foreach(println)

8)sample(withReplacement, fraction, seed)

以指定的随机种子随机抽样出数量为fraction的数据,withReplacement表示是抽出的数据是否放回,true为有放回的抽样,false为无放回的抽样,seed用于指定随机数生成器种子。

scala

val config = new SparkConf().setMaster("local[*]").setAppName("WordCount")

val sc = new SparkContext(config)

val listRDD = sc.makeRDD(1 to 16)

val sampleRDD = listRDD.sample(false, 0.4, 1)

sampleRDD.collect().foreach(println)

9)distinct([numTasks]))

对源RDD进行去重后返回一个新的RDD。默认情况下,只有8个并行任务来操作,但是可以传入一个可选的numTasks参数改变它。

scala

val config = new SparkConf().setMaster("local[*]").setAppName("WordCount")

val sc = new SparkContext(config)

val listRDD = sc.makeRDD(1 to 16)

val distinctRDD = listRDD.distinct()

distinctRDD.collect().foreach(println)

10)coalesce(numPartitions)

缩减分区数,用于大数据集过滤后,提高小数据集的执行效率。

scala

val config = new SparkConf().setMaster("local[*]").setAppName("WordCount")

val sc = new SparkContext(config)

val listRDD = sc.makeRDD(1 to 16, 4)

println("缩减分区前=" + listRDD.partitions.size)

val coalesceRDD = listRDD.coalesce(3)

println("缩减分区后=" + coalesceRDD.partitions.size)

11) repartition(numPartitions)

根据分区数,重新通过网络随机洗牌所有数据。

scala

val config = new SparkConf().setMaster("local[*]").setAppName("WordCount")

val sc = new SparkContext(config)

val listRDD = sc.makeRDD(1 to 16, 4)

val repartitionRDD = listRDD.repartition(2)

repartitionRDD.collect().foreach(println)

coalesce和repartition的区别

    1. coalesce重新分区,可以选择是否进行shuffle过程。由参数shuffle: Boolean = false/true决定。
    1. repartition实际上是调用的coalesce,默认是进行shuffle的。源码如下:

scala

def repartition(numPartitions: Int)(implicit ord: Ordering[T] = null): RDD[T] = withScope {
 coalesce(numPartitions, shuffle = true)
}

12)sortBy(func,[ascending], [numTasks])

使用func先对数据进行处理,按照处理后的数据比较结果排序,默认为正序(从小到大)。

scala

val config = new SparkConf().setMaster("local[*]").setAppName("WordCount")

val sc = new SparkContext(config)

val listRDD = sc.makeRDD(1 to 16, 4)

//安装除以2余数来排序
val sortByRDD = listRDD.sortBy(x=>x%2)

sortByRDD.collect().foreach(println)

13)pipe(command, [envVars])

管道,针对每个分区,都执行一个shell脚本,返回输出的RDD。

2、双Value类型

1)union(otherDataset)

对源RDD和参数RDD求并集后返回一个新的RDD。

2)subtract (otherDataset)

计算差的一种函数,去除两个RDD中相同的元素,不同的RDD将保留下来。

3)intersection(otherDataset)

对源RDD和参数RDD求交集后返回一个新的RDD。

4)cartesian(otherDataset)

笛卡尔积(尽量避免使用)

5)zip(otherDataset)

将两个RDD组合成Key/Value形式的RDD,这里默认两个RDD的partition数量以及元素数量都相同,否则会抛出异常。

3、Key-Value类型

1)partitionBy

对pairRDD进行分区操作,如果原有的partionRDD和现有的partionRDD是一致的话就不进行分区, 否则会生成ShuffleRDD,即会产生shuffle过程。

scala

def main(args: Array[String]): Unit = {

  val config = new SparkConf().setMaster("local[*]").setAppName("WordCount")

  val sc = new SparkContext(config)

  val listRDD = sc.makeRDD(List(("a",1),("b",2),("c",3)))

  val partRDD = listRDD.partitionBy(new MyPartitioner(3))

  partRDD.saveAsTextFile("output")

}

//自定义
class MyPartitioner(partitions:Int) extends Partitioner{
  //设置分区数目
  override def numPartitions: Int = {
    partitions
  }

  //设置数据存放位置
  override def getPartition(key: Any): Int = {
    1
  }
}

2)groupByKey

groupByKey也是对每个key进行操作,但只生成一个sequence

scala

val config = new SparkConf().setMaster("local[*]").setAppName("WordCount")

val sc = new SparkContext(config)

val words = Array("one", "two", "two", "three", "three", "three")

val wordPairsRDD = sc.parallelize(words).map(word => (word, 1))

val group = wordPairsRDD.groupByKey()

//将相同key对应值聚合到一个sequence中
group.collect().foreach(println)
//计算相同key对应值的相加结果
group.map(t=>(t._1,t._2.sum)).foreach(println)

3)reduceByKey(func, [numTasks])

在一个(K,V)的RDD上调用,返回一个(K,V)的RDD,使用指定的reduce函数,将相同key的值聚合到一起,reduce任务的个数可以通过第二个可选的参数来设置。

reduceByKey和groupByKey的区别

    1. reduceByKey:按照key进行聚合,在shuffle之前有combine(预聚合)操作,返回结果是RDD[k,v].
    1. groupByKey:按照key进行分组,直接进行shuffle。
    1. 开发指导:reduceByKey比groupByKey快,建议使用。但是需要注意是否会影响业务逻辑。

4)aggregateByKey

参数:(zeroValue:U,[partitioner: Partitioner]) (seqOp: (U, V) => U,combOp: (U, U) => U)

    1. 作用:在kv对的RDD中,,按key将value进行分组合并,合并时,将每个value和初始值作为seq函数的参数,进行计算,返回的结果作为一个新的kv对,然后再将结果按照key进行合并,最后将每个分组的value传递给combine函数进行计算(先将前两个value进行计算,将返回结果和下一个value传给combine函数,以此类推),将key与计算结果作为一个新的kv对输出。
    1. 参数描述:

(1)zeroValue:给每一个分区中的每一个key一个初始值; (2)seqOp:函数用于在每一个分区中用初始值逐步迭代value; (3)combOp:函数用于合并每个分区中的结果。

创建一个pairRDD,取出每个分区相同key对应值的最大值,然后相加

scala

val config = new SparkConf().setMaster("local[*]").setAppName("WordCount")

val sc = new SparkContext(config)

val rdd = sc.parallelize(List(("a", 3), ("a", 2), ("c", 4), ("b", 3), ("c", 6), ("c", 8)), 2)

rdd.glom().collect().foreach(println)

val value = rdd.aggregateByKey(0)(math.max(_, _), _ + _)

value.collect().foreach(println)

5)foldByKey

参数:(zeroValue: V)(func: (V, V) => V): RDD[(K, V)]

    1. 作用:aggregateByKey的简化操作,seqop和combop相同
    1. 需求:创建一个pairRDD,计算相同key对应值的相加结果

scala

val agg = rdd.foldByKey(0)(_+_)

6)combineByKey[C]

参数:(createCombiner: V => C, mergeValue: (C, V) => C, mergeCombiners: (C, C) => C)

    1. 作用:对相同K,把V合并成一个集合。
    1. 参数描述:

    (1)createCombiner: combineByKey() 会遍历分区中的所有元素,因此每个元素的键要么还没有遇到过,要么就和之前的某个元素的键相同。如果这是一个新的元素,combineByKey()会使用一个叫作createCombiner()的函数来创建那个键对应的累加器的初始值

(2)mergeValue: 如果这是一个在处理当前分区之前已经遇到的键,它会使用mergeValue()方法将该键的累加器对应的当前值与这个新的值进行合并 (3)mergeCombiners: 由于每个分区都是独立处理的, 因此对于同一个键可以有多个累加器。如果有两个或者更多的分区都有对应同一个键的累加器, 就需要使用用户提供的 mergeCombiners() 方法将各个分区的结果进行合并。

scala

val config = new SparkConf().setMaster("local[*]").setAppName("WordCount")

val sc = new SparkContext(config)

val input = sc.parallelize(Array(("a", 88), ("b", 95), ("a", 91), ("b", 93), ("a", 95), ("b", 98)),2)

val value = input.combineByKey(x => x, (x: Int, y: Int) => x + y, (x: Int, y: Int) => x + y)

value.collect().foreach(println)

7)sortByKey([ascending], [numTasks])

作用:在一个(K,V)的RDD上调用,K必须实现Ordered接口,返回一个按照key进行排序的(K,V)的RDD

scala

val input = sc.parallelize(Array(("a", 88), ("b", 95), ("a", 91), ("b", 93), ("a", 95), ("b", 98)),2)
val value1 = input.sortByKey(true)

8)mapValues

针对于(K,V)形式的类型只对V进行操作

scala

val config = new SparkConf().setMaster("local[*]").setAppName("WordCount")
val sc = new SparkContext(config)
val input = sc.parallelize(Array(("a", 88), ("b", 95), ("a", 91), ("b", 93), ("a", 95), ("b", 98)), 2)
val value1 = input.mapValues(_ + 100)
value1.collect().foreach(println)

9)join(otherDataset, [numTasks])

在类型为(K,V)和(K,W)的RDD上调用,返回一个相同key对应的所有元素对在一起的(K,(V,W))的RDD

10)cogroup(otherDataset, [numTasks])

在类型为(K,V)和(K,W)的RDD上调用,返回一个(K,(Iterable,Iterable))类型的RDD

三、Action(行动算子)

1)reduce(func)

通过func函数聚集RDD中的所有元素,先聚合分区内数据,再聚合分区间数据。

2)collect()

在驱动程序中,以数组的形式返回数据集的所有元素。

3) count()

返回RDD中元素的个数

4)first()

返回RDD中的第一个元素

5)take(n)

返回一个由RDD的前n个元素组成的数组

6)takeOrdered(n)

返回该RDD排序后的前n个元素组成的数组

7)aggregate(num)(func1)(func2)

    1. 参数:(zeroValue: U)(seqOp: (U, T) ⇒ U, combOp: (U, U) ⇒ U)
    1. 作用:aggregate函数将每个分区里面的元素通过seqOp和初始值进行聚合,然后用combine函数将每个分区的结果和初始值(zeroValue)进行combine操作。这个函数最终返回的类型不需要和RDD中元素类型一致。

8)fold(num)(func)

作用:折叠操作,aggregate的简化操作,seqop和combop一样。

9)saveAsTextFile(path)

将数据集的元素以textfile的形式保存到HDFS文件系统或者其他支持的文件系统,对于每个元素,Spark将会调用toString方法,将它装换为文件中的文本

10)saveAsSequenceFile(path)

将数据集中的元素以Hadoop sequencefile的格式保存到指定的目录下,可以使HDFS或者其他Hadoop支持的文件系统。

11)saveAsObjectFile(path)

用于将RDD中的元素序列化成对象,存储到文件中。

12) countByKey()

针对(K,V)类型的RDD,返回一个(K,Int)的map,表示每一个key对应的元素个数。

scala

val rdd = sc.parallelize(List((1,3),(1,2),(1,4),(2,3),(3,6),(3,8)),3)

val total = rdd.countByKey()

println(total)

//打印结果
Map(3 -> 2, 1 -> 3, 2 -> 1)

13) foreach(func)

在数据集的每一个元素上,运行函数func进行更新。

四、RDD依赖关系

1)Lineage

RDD只支持粗粒度转换,即在大量记录上执行的单个操作。将创建RDD的一系列Lineage(血统)记录下来,以便恢复丢失的分区。RDD的Lineage会记录RDD的元数据信息和转换行为,当该RDD的部分分区数据丢失时,它可以根据这些信息来重新运算和恢复丢失的数据分区。

2)窄依赖

窄依赖指的是每一个父RDD的Partition最多被子RDD的一个Partition使用.

3)宽依赖

宽依赖指的是多个子RDD的Partition会依赖同一个父RDD的Partition,会引起shuffle.

4)DAG(Directed Acyclic Graph)

DAG(Directed Acyclic Graph)叫做有向无环图,原始的RDD通过一系列的转换就就形成了DAG,根据RDD之间的依赖关系的不同将DAG划分成不同的Stage,对于窄依赖,partition的转换处理在Stage中完成计算。对于宽依赖,由于有Shuffle的存在,只能在parent RDD处理完成后,才能开始接下来的计算,因此宽依赖是划分Stage的依据

五、累加器

累加器用来对信息进行聚合,通常在向 Spark传递函数时,比如使用 map() 函数或者用 filter() 传条件时,可以使用驱动器程序中定义的变量,但是集群中运行的每个任务都会得到这些变量的一份新的副本,更新这些副本的值也不会影响驱动器中的对应变量。如果我们想实现所有分片处理时更新共享变量的功能,那么累加器可以实现我们想要的效果。

scala

val config = new SparkConf().setMaster("local[*]").setAppName("WordCount")

val sc = new SparkContext(config)

val rdd: RDD[Int] = sc.makeRDD(List(1, 2, 3, 4), 2)

//创建累加器对象
val accumulator: LongAccumulator = sc.longAccumulator

rdd.foreach {
  case i => {
    //执行累加器的累加功能
    accumulator.add(i)
  }
}
//获取累加器保存的值
println(accumulator.value)

自定义累加器

自定义累加器类型的功能在1.X版本中就已经提供了,但是使用起来比较麻烦,在2.0版本后,累加器的易用性有了较大的改进,而且官方还提供了一个新的抽象类:AccumulatorV2来提供更加友好的自定义类型累加器的实现方式。实现自定义类型累加器需要继承AccumulatorV2并至少覆写下例中出现的方法。

六、广播变量(调优策略)

广播变量用来高效分发较大的对象。向所有工作节点发送一个较大的只读值,以供一个或多个Spark操作使用。比如,如果你的应用需要向所有节点发送一个较大的只读查询表,甚至是机器学习算法中的一个很大的特征向量,广播变量用起来都很顺手。 在多个并行操作中使用同一个变量,但是 Spark会为每个任务分别发送。

使用广播变量的过程如下:

  • (1) 通过对一个类型 T 的对象调用 SparkContext.broadcast 创建出一个 Broadcast[T] 对象。 任何可序列化的类型都可以这么实现。
  • (2) 通过 value 属性访问该对象的值(在 Java 中为 value() 方法)。
  • (3) 变量只会被发到各个节点一次,应作为只读值处理(修改这个值不会影响到别的节点)。

本文参与腾讯云自媒体分享计划,欢迎正在阅读的你也加入,一起分享。

我来说两句

0 条评论
登录 后参与评论

相关文章

  • 【Spark】Spark SQL原理、编译、配置及运行方式详述

    (1)在Hadoop中运行SQL的工具 在Hadoop中运行SQL的工具有Hive、Impala、Apache Drill、Presto、Spark SQL等...

    魏晓蕾
  • BigData--分布式流数据流引擎Apache Flink

    MiChong
  • 基于云原生的大数据实时分析方案实践

    徐蓓,腾讯云容器专家工程师,10年研发经验,7年云计算领域经验。负责腾讯云 TKE 大数据云原生、离在线混部、Serverless 架构与研发。 1 方案介绍...

    腾讯云原生
  • Spark SQL 快速入门系列(8) | | Hive与Spark SQL的读写操作

      包含 Hive 支持的 Spark SQL 可以支持 Hive 表访问、UDF (用户自定义函数)以及 Hive 查询语言(HiveQL/HQL)等。需要强...

    不温卜火
  • 大数据高速计算引擎Spark

    从狭义的角度上看:Hadoop是一个分布式框架,由存储、资源调度、计算三部分组 成; Spark是一个分布式计算引擎,由 Scala 语言编写的计算框架,基于内...

    用户2337871
  • 超快!大数据分析引擎ClickHouse

    ClickHouse是一个用于联机分析(OLAP)的列式数据库管理系统(DBMS)。

    凹谷
  • 大数据分析查询引擎Impala

    准实时分析系统Impala,提供SQL语义,能够为存储在Hadoop的HDFS和Hbase中的PB级大数据提供快速、交互式的SQL查询。传统仓库查询工具Hive...

    凹谷
  • 干货|流批一体Hudi近实时数仓实践

    传统意义上的数据集市主要处理T+1的数据。随着互联网的发展,当前越来越多的业务场景对于数据时效性提出了更高的要求,以便及时快速地进行数据分析和业务决策,比如依托...

    大数据老哥
  • [大数据架构 ]Apache大数据项目目录

    在使用BigData大约8年以上之后,我遇到了大量的项目。Esp Apache的运动对于BigData域非常强大。每个人都会提出一个针对特定解决方案的项目。但是...

    首席架构师智库
  • 通过扩展 Spark SQL ,打造自己的大数据分析引擎

    Spark SQL 的 Catalyst ,这部分真的很有意思,值得去仔细研究一番,今天先来说说Spark的一些扩展机制吧,上一次写Spark,对其SQL的解析...

    麒思妙想
  • 利用Spark实现Oracle到Hive的历史数据同步

    和上一篇文章Spark通过修改DataFrame的schema给表字段添加注释一样,通过Spark将关系型数据库(以Oracle为例)的表同步的Hive,这里讲...

    董可伦
  • 大数据常用技术栈

    提起大数据,不得不提由IBM提出的关于大数据的5V特点:Volume(大量)、Velocity(高速)、Variety(多样)、Value(低价值密度)、Ver...

    大数据学习与分享
  • 大数据常用技术栈

    提起大数据,不得不提由IBM提出的关于大数据的5V特点:Volume(大量)、Velocity(高速)、Variety(多样)、Value(低价值密度)、Ver...

    大数据学习与分享
  • spark-submit报错: java.sql.SQLException:No suitable driver

    转载请务必注明原创地址为:http://dongkelun.com/2018/05/06/sparkSubmitException/

    董可伦
  • 大数据计算引擎,选 Flink 还是 Spark?

    我是 2018 年 6 月加入公司,一直负责监控平台的告警系统。之后,我们的整个监控平台架构中途换过两次,其中一次架构发生了巨大的变化。我们监控告警平台最早的架...

    CSDN技术头条
  • 2021年大数据Spark(五十一):Structured Streaming 物联网设备数据分析

    在物联网时代,大量的感知器每天都在收集并产生着涉及各个领域的数据。物联网提供源源不断的数据流,使实时数据分析成为分析数据的理想工具。

    Lanson
  • 使用 Iceberg on Kubernetes 打造新一代云原生数据湖

    作者徐蓓,腾讯云容器专家工程师,10年研发经验,7年云计算领域经验。负责腾讯云 TKE 大数据云原生、离在线混部、Serverless 架构与研发。 背景 大...

    腾讯云原生
  • 2021年大数据Spark(三十三):SparkSQL分布式SQL引擎

    SparkSQL模块从Hive框架衍生发展而来,所以Hive提供的所有功能(数据分析交互式方式)都支持,文档:http://spark.apache.org/d...

    Lanson
  • hive查询报错:java.io.IOException:org.apache.parquet.io.ParquetDecodingException

    转载请务必注明原创地址为:https://dongkelun.com/2018/05/20/hiveQueryException/

    董可伦

扫码关注云+社区

领取腾讯云代金券