spark笔记

川长思鸟来 2022-05-17 05:10 223阅读 0赞

第一章 Spark介绍  
1. Spark简介

1.1 定义:
            Spark是一个快速且通用的集群计算平台.
    
        1.2 Spark特点:
            Spark扩充了流行的Mapreduce计算模型.
            Spark是基于内存的计算
            Spark是通用的:Spark的设计容纳了其他分布式系统拥有的功能,批处理,迭代式计算,交互查询(hive)和流处理(Stream)
            大大降低了维护成本
            Spark是高度开放的:
                Spark提供了Python,java,Scala,SQL的API和丰富的内置库

2. Spark生态介绍  
2.1 起源:  
最初诞生于2009年,最初是基于Hadoop Mapreduce,后来发现Mapreduce在迭代式计算和交互式上低效,引入了内存存储,从此产生了巨大的区别.  
2010年三月份Spark开源,2011年AMP实验室在Spark上开发高级组件,像Spark Streaming  
2013年转移到Apache下,不就便成为顶级项目  
2.2 Spark组件:  
![这里写图片描述][70]  
2.2.1 SparkCore:  
包含了Spark的基本功能,包含任务调度,内存管理,容错机制等  
内部定义了RDDS(弹性分布式数据集)  
提供了很多APIs来创建和操作这些RDDS  
应用场景,为其他组件提供底层的服务

2.2.2 Spark SQL:
                是Spark处理结构化数据的库,像Hive SQL,Nysql一样.
                应用场景,企业中用来做报表统计
    
            2.2.3 Spark Streaming:
                是实时数据流处理组件,类似Storm.
                Spark Streaming提供了API来操作实时流数据.
                应用场景,企业中用来从kafka接收数据做实时统计
    
            2.2.4 Milb:
                一个包含通用机器学习功能的包,Machine learning lib.
                包含分类,聚类,回归等,还包括模型评估和数据导入.
                Mlib提供的上面这些方法,都支持集群上的横向扩展.
                应用场景,机器学习
    
            2.2.5 Graphx:
                是处理图的库(例如,社交网络图),并进行图的并行计算.
                像Spark Streaming,Spark SQL一样,它也继承了RDD API.
                它提供了各种图操作,和常用的算法,例如RangeRank算法.
                应用场景,图计算
    
            2.2.6 Cluster Managers:
                就是集群管理,Spark自带一个集群管理是单独调度器.
                常见的集群管理包括Hadoop YARN,Apache Mesos
    
            2.2.7 紧密集成的优点:
                Spark底层又花了,基于Spark底层的组件,也得到了相应的优化.
                紧密集成,节省了各个组件组合使用时的部署,测试等时间.
                向Spark增加新的组件时,其他组件,可立刻享用新组件的功能.

3. spark与Hadoop的比较

3.1 Hadoop的应用场景
            离线处理
            对时效性要求不高的数据
    
        3.2 Spark应用场景
            时效性要求高的场景(基于内存的)
            机器学习等领域
    
        3.3 比较
            Doug Cutting的观点:
               这是生态系统,每个组件都有其作用,各善其职即可
               Spark不具有HDFS的存储能力,要借助HDFS等持久化数据.

第二章 Spark的下载和安装

1. Spark的安装:
    
        wget https://archive.apache.org/dist/spark/spark-2.0.2/spark-2.0.2-bin-hadoop2.7.tgz
        解压缩即可
    
    2. Spark目录:
    
        bin包含用来和Spark交互的可执行文件,如Spark shell
        core ,streaming,python...包含主要组件的源代码.
        examples包含一些单机Spark job,你可以研究和运行这些例子
    
    3. Spark的Shell:
    
        Spark的shell使你能够处理分布在集群上的数据
        Spark把数据加载到节点的内存中,因此分布式处理可在秒级完成.
        快速使迭代式计算,实时查询.分析一般能够在shells中完成
        Spark提供了Python shells和Scala shells
    
    4. Scala Shell:
    
        在bin/spark-shell目录下
        例子:(读取本机文件,并对文件内容做统计)
            var lines = sc.textFile("../../testFile/helloSpark")
            lines.count()  统计数据的总条数
    
            修改日志级别:
                log4j.rootCategory=WARN,console
                进入到spark中的conf文件夹中然后

第三章 开发第一个Spark程序

1. Spark的开发环境搭建
    
        需要安装scala的插件,在idea中进行安装是非常方便的.
    
    2. 开发第一个Spark程序
    
      2.1 配置ssh无密登录:
            ssh-keygen
            .ssh目录下 cat xxx_rsa.pub > authorized_keys
            chmod 600 authorized_keys
    
            步骤:
               root目录下输入 ssh-keygen 然后一直回车即可
               然后cd .ssh进入到改目录下,没有的话可以创建
                vim authorized_keys
               继而执行:
                cat id_rsa.pub > authorized_keys
               最后一步修改权限
               chmod 600 authorized_keys
               成功结果如下图:

![这里写图片描述][70 1]  
2.2 程序的开发 WordCount开发:

2.2.1 开发步骤:
    
                创建一个Spark Context   (首先在build.sbt文件中加入libraryDependencies ++= Seq("org.apache.spark" %% "spark-core" % "1.6.2")

然后刷新项目)  
加载数据  
把每一行分割成单词  
转换成pairs并且计数

2.2.2 打包过程:
                  在项目中点击File->Project Structure->Artifacts然后点击加号,添加jar->from module 选择项目,然后类名就可以了
    
                  点击ok之后,就可以点击build->build Artifacts->build
    
           2.2.3 启动集群:
    
                   启动master   ./sbin/start-master.sh
                   启动worker   ./bin/spark-class
                   提交作业      ./bin/spark-submit

第四章 Rdds  
1. RDDs介绍  
![这里写图片描述][70 2]  
Driver program:  
包含程序的main()方法,RDDs的定义和操作.它管理很多节点,我们称作executors.如下图:  
SparkContext:  
Driver programs通过SparkContext对象访问Spark.  
SparkContext对象代表和一个集群的连接  
在Shell中SparkContext自动创建好了,就是sc.

RDDs介绍:
    
           Resilient distributed datasets(弹性分布式数据集,简写RDDs)
           这些RDDs,并行的分布在整个集群中.
           RDDs是Spark分发数据和计算的基础抽象类
           RDDs是一个不可改变的分布式集合对象,在Spark中,所有的计算都是通过RDDs的创建,转换,操作的.
           一个RDDs内部由许多partitions(分片)组成.
           分片:
               每个分片包括一部分数据,partitions可在集群不同节点计算.
               分片是Spark并行处理的单元,Spark顺序的,并行的处理分片
    
       RDDs的创建方法:
    
            把一个存在的集合传给SparkContext的parallelize()方法,测试用
            val rdd = sc.parallelize(Array(1,2,3,4),4)
            第一个参数:
                待并行化处理的集合,第2个参数:分区个数
            加载外部数据集:
                val rddText = sc.textFile("helloSpark.txt")  //此处可以加载不同位置的,需要处理的文件
    
       scala的基础知识:
    
            scala的匿名函数和类型推断:
            lines.filter(line => line.contains("world"))
            定义一个匿名函数,接受一个参数line,过滤含有world字符串的数据
            使用line这个String类型变量上的contains方法,并且返回结果
            line的类型不需要指定,能够推断出来.

2. Rdds基本操作之Transformations

2.1 逐元素Transformations
    
            eg:  map()用法
            map()接收函数,把函数应用到RDD的每一个元素,返回新RDD,
    
              val lines2 = lines.map(world => (world,1))  在world后面加一个后缀(world,1)
    
            eg: filter() 过滤满足条件的数据
    
              val lines3 = lines.filter(world => world.contains("hello")) //过滤包含hello的数据
              lines3.foreach(println)
    
            eg: flatMap()
    
                对每个输入元素,输出多个输出元素.
                flat压扁的意思,将RDD中元素压扁后返回一个新的RDD.
    
                val lines = inputs.flatMap(line => line.split(" ")) //以空格进行分割,实际就是去掉空格

3. RDDS基本操作之Transfromations  
3.1 集合运算:  
RDDs支持数学集合的运算,例如并集,交集计算.  
第一步:  
创建数据: val rdd1 = sc.parallelize(Array(“coffe”,”coffe”,”panda”,”monkey”,”tea”))  
val rdd2 = sc.parallelize(Array(“coffe”,”monkey”,”kitty”))  
操作:  
val rdd\_distinct = rdd1.distinct() //去重操作  
val rdd\_union = rdd1.union(rdd2) //并集,把两者合并  
val rdd\_inter = rdd1.intersection(rdd2) //交集  
val rdd\_sub = rdd1.subtract(rdd2) //rdd1 有的,rdd2没有的

4. RDD基本操作之Action
    
         在RDD上计算出来一个结果,把结果返回给driver program或保存在文件系统,比如count(),save函数
         下面是常用的action,下面就介绍几个常用的:

![这里写图片描述][70 3]  
4.1 reduce():  
接收一个函数,作用在RDD两个类型相同的元素上,返回新元素.  
可以实现,RDD中元素的累加,计数,和其他类型的聚集操作  
eg: (只要类型相同就执行累加操作)  
val rdd =sc.parallelize(Array(1,2,3,3))  
scala> rdd.collect()  
res1: Array\[Int\] = Array(1, 2, 3, 3)  
scala> rdd.reduce((x,y) => (x+y))  
res2: Int = 9  
4.2 Collect():  
遍历整个RDD,向driver program返回RDD的内容  
需要单机内存能够容纳下(因为数据拷贝给driver,测试使用)  
大数据的时候,使用saveAsTextFile() action等操作.

4.3 take(n) :
             返回RDD的n个元素(同时尝试访问最少的partitions).
             返回结果是无序的,测试使用.
             eg:
                scala> rdd.take(2)
                res3: Array[Int] = Array(1, 2)
    
                scala> rdd.take(3)
                res4: Array[Int] = Array(1, 2, 3)
    
         4.4 top():
             排序(根据RDD中数据的比较器)
             eg:
               scala> rdd.take(3)
               res4: Array[Int] = Array(1, 2, 3)
               scala> rdd.top(1)
               res5: Array[Int] = Array(3)
               scala> rdd.top(2)
               res6: Array[Int] = Array(3, 3)
               scala> rdd.top(3)
               res7: Array[Int] = Array(3, 3, 2)
    
         4.5 foreach():
            计算RDD中的每个元素,但不返回到本地,可以配合println()友好的打印出数据
    
    5. RDDs的特性
    
        5.1 RDDs的血统关系图:

![这里写图片描述][70 4]  
5.2 延迟计算(Lazy Evaluation)  
Spark对RDDs的计算是,他们第一次使用action操作的时候,这种方式在处理大数据的时候特别有用,可以减少数据的传输.  
Spark内部记录metadata表明transformations操作已经被响应了,加载数据也是延迟计算,数据只有在必要的时候,才会被加载进去.

5.3 RDD.persist(): (缓存)
            默认每次在RDDs上面进行action操作时,Spark都重新计算RDDs,如果想重复利用一个RDD,可以使用RDD.persist()
            unpersist()方法,是从缓存中移除数据
            缓存级别:

![这里写图片描述][70 5]  
![这里写图片描述][70 6]  
6. KeyValue对RDDS(一)

6.1 创建KeyValue对RDDs:
            使用map()函数,返回key/value对
              eg:
                    scala> rdd.foreach(println)
                    hello spark
                    hello world
                    hello !
    
                   val rdd2 = rdd.map(line => (line.split(" ")(0),line))
    
                   scala> rdd2.foreach(println)
                   (hello,hello spark)
                   (hello,hello world)
                   (hello,hello !)
    
        具体操作图示:
    
    
    7. KeyValue对RDDS(二)
    
        7.1 combineByKey()函数:
              (createCombiner,mergeValue,mergeCombiners,partitioner)
              最常用的基于Key的聚合函数,返回的类型可以与输入类型不一样
              许多基于key的聚合函数都用到了它,像groupByKey()
    
              遍历partition中的元素,元素的key,要么之前见过的,要么不是.
              如果是新元素,使用我们提供的createCombiner()函数
              如果是这个partition中已经存在的key,就会使用mergeValue()函数
              合计每个partition的结果的时候,使用mergeCombiners()函数.

第五章 课程总结

[70]: /images/20220517/9211b500f138415b9531b6242cb1bf8e.png
[70 1]: /images/20220517/6e4e403c9a4a44faae3ab027d192ccdb.png
[70 2]: /images/20220517/aed40bd5ffa04bad8aa5f82983b58100.png
[70 3]: /images/20220517/0ee6fd9049374211b1e84fb2921cbc4e.png
[70 4]: /images/20220517/3123bea26cb547888dac32460524b6b3.png
[70 5]: /images/20220517/45e35cabe91949aebf8d8a969c3b8d8e.png
[70 6]: /images/20220517/d6085212a7a649dbaf4efb9702ac354f.png