Spark 基础知识-03

目录

2、RDD的算子

七、RDD的持久化

1、RDD的数据是过程数据

2、RDD的缓存技术

3、RDD的CheckPoint技术

八、案例-搜索引擎日志分析案例

1、案例介绍

2、业务需求

3、jieba库的入门使用

4、功能实现

5、将代码提交到集群中运行

九、共享变量

1、广播变量

2、累加器

3、综合案例

---

衔接Spark 基础知识-02

2、RDD的算子

        （15）countByKey算子-A

        功能：针对kv型RDD统计key出现的次数

        

        （17）collect算子-A

        功能：将RDD各个分区内的数据统一收集到Driver中，返回一个list对象。注意collect算子适用于数据集不大的情况，如果数据太大，所有数据都会收集到Driver的内存中，会导致内存使用量增加，甚至溢出。

        （18）reduce算子-A

        功能：对RDD数据按照传入的逻辑进行聚合，返回值受传入函数的影响。

        

        

        （19）fold算子-A-了解

        功能：与reduce算子一样，接收传入逻辑进行聚合，与reduce不同的是聚合是带有初始值的。

         如果数据有分区的话，这个初始值在分区内和分区间都会生效：

        

        （20）first算子-A

        功能：取出RDD的第一个元素

        

        （21）take算子-A

        功能：取RDD的前N个元素，组成list返回

        （22）top算子-A

        功能：对RDD数据集进行降序排序，取前N个，返回list

        

        （23）count算子-A

        功能：计算RDD有多少条数据，返回一个数字

        

        （24）takeSample算子-A

        功能：随机抽样RDD的数据（由于Spark是处理海量数据的框架如果直接使用collect可能会让Driver内存溢出，因此可以使用随机采样用部分估计整体）

 

 

        （25）takeOrdered算子-A

        功能：对RDD进行排序取前N个，比top算子功能强，top算子只能进行降序排序

        （26）foreach算子-A

        功能：对RDD的每个元素执行所提供的逻辑函数，类似与map，但是它没有返回值。因此可以将foreach当作没有返回值的map算子。

        

 

 

        注意foreach算子的执行是由Executor直接输出的，不像collect一样需要先将分区的数据收集到Driver中然后在输出，因此foreach要比collect算子执行效率快。针对没有返回值的场景使用较多，比如向文件中写数据，会并行写入，速度快。

        （27）saveAsTextFile算子-A

        功能：将RDD的数据写入到文本文件中，支持本地文件和hdfs等文件系统

        

        由于当前pycharm使用的是远程解析器，代码本质上是在linux系统上执行的，因此数据也保存在linux系统上：（保存至本地目录不存在会自动创建）

        如果想要在Windows本机上获取数据，需要手动同步一下：

 

 

         将RDD数据写入到hdfs文件系统中，注意在执行代码之前需要保证hdfs中有相应的目录（output目录已存在，out1不能存在），以及修改权限为777。这是为了避免使用本地用户xxxxxx去操作hdfs的文件权限不足的问题。

        注意：与foreach算子一样，saveAsTextFile算子的执行也是由Executor直接输出的，也就是说上述part00000，part00001，part00002是并行生成的。

        总结：所有的Action算子中只有foreach和saveAsTextFile两个算子是Executor直接执行的，其余的Action算子都会将结果先发送给Driver，然后由Driver统一输出。

        （28）mapPartitions算子-T

        功能：与map算子功能一样，但是运算过程不同：

        因此上图中mapPartitions算子中的函数不能像map算子一样lambda x: x*10来写，对于map来说传入过来的x分别是1，2，3，4，5，6，7，8，9因此可以直接计算，对于mapPartitions来说传入的x分别是[1,2,3]，[4,5,6]，[7,8,9]，列表怎么能和10相乘计算呢！！

        （29）foreachPartition算子-A

        功能：与foreach功能一样，就是一次处理整个分区数据，foreachPartition本质上就是没有返回值的mapPartitions

        （30）partitionBy算子-T

        功能：对RDD进行自定义分区操作

        例如参数1传入5，则参数2的函数返回值必须是0，1，2，3，4，相同值的数据放在一个分区内。

        （31）repartition算子-T

        功能：对RDD的数据进行重新分区，无法重新指定分区规则，只能修改分区数

        注意：该算子不建议多用，且使用时要慎重，即使使用也是减少分区，因为如果更改了分区，会影响到并行计算的过程。增加分区极大可能会导致shuffle产生，导致性能下降。

        （32）coalesce算子-T

        功能：对RDD的数据进行重新分区，无法重新指定分区规则，只能修改分区数，与repartition功能一样但是唯一不同的时，coalesce算子在增加分区时有安全机制。目的是防止误操作。

        在减少分区时与repartition算子一样，在增加分区时，由于极大可能会导致shuffle产生，因此必须在coalesce中添加一个参数允许执行shuffle操作：shuffle = True，此时增加分区才会生效：

 

        补充：面试题groupByKey与reduceByKey的区别

        在功能上：groupByKey仅仅有分组功能；reduceByKey有分组+聚合的功能

        如果对数据执行分组+聚合操作，那么这两个算子的效率差距是很大的：reduceByKey的执行性能远大于groupByKey + 聚合算子：

         由于reduceByKey自带聚合逻辑，所以在执行过程中会先在分区内做预聚合，然后在分组（shuffle），最后在做最终的聚合：

         reduceByKey性能比groupByKey好的最大原因就是提前进行了聚合操作，使得在shuffle时，被shuffle的数据量大量减少，从而提升效率。

七、RDD的持久化

1、RDD的数据是过程数据

        注意：在默认情况下Spark会自动记录RDD之间的Lineage（血统），以便后续恢复丢失的分区（RDD的容错机制）。RDD的Lineage会记录RDD的元数据信息和转换行为（血缘关系），但是其不会记录RDD的数据本身。因此，各个RDD之间进行相互迭代计算（Transformation的转换）时，新的RDD生成，代表老的RDD消失。RDD的数据是过程数据，只在处理的过程中存在，一旦处理完成就消失了。该特性可以最大化利用资源，老旧RDD没用了就从内中中清理，给后续的计算腾出内存空间。

        如上图，rdd1，2，3被调用两次，在第一次使用完之后rdd1，2，3就不存在了，通过rdd3生成rdd5的时候，需要基于rdd的血缘关系（Lineage记录），从rdd1重新执行构建出rdd3供rdd5使用，这看起来是不是很不好，有点浪费资源，明明已经将rdd1，2，3生成过一次了，后续继续使用需要重新生成。由于RDD默认情况下不保存数据，这就导致当重复使用RDD时只能通过依赖上游的血缘关系恢复当前的RDD，然后在执行接下来的操作，这就很麻烦。此时就需要RDD的持久化技术（RDD缓存和检查点技术）来解决这个问题。

2、RDD的缓存技术

        RDD的缓存技术：Spark提供了缓存API算子，可以让我们通过调用API将指定的RDD数据保留在内存或者磁盘上，这样后续继续使用时就不需要重新生成了。针对上述问题我们可以将RDD3的数据保留在内存或磁盘上，生成RDD5时不需要重新构建RDD3，直接使用已经保存的RDD3即可。

        注意：上述的缓存技术只保存RDD的数据本身。RDD的缓存与RDD血缘关系的存储（Lineage）是不同的概念，存储的位置也不同，无论用不用cache，persist缓存技术，Spark都会将血缘关系和其他元数据永久保存在一个特定的位置。

        注意使用persist需要导入StorageLevel包：from pyspark.storagelevel import StorageLevel

        通过time.sleep(100000)将程序卡住这就可以看4040端口了，总共有两个job被执行，因为使用了两次collect。

        观察20行子job的DAG图可以看到如果不使用缓存技术，RDD1，2，3会重新生成一次：

        将RDD3的数据放入缓存中以后，再次打开相应的DAG图可以看到绿点就是已经缓存的RDD3，后续的执行都是从绿点开始：

        调用缓存的内部过程：

        缓存技术的问题：缓存在设计上被认为是不安全的，因为存在缓存丢失问题：首先如果数据是存储在内存中，此时出现断电（内存被清空）、内存不足（有其他任务需要使用内存时，由于内存不足，计算机会自动将内存进行清理）的情况，都会导致缓存丢失。如果数据是存储在磁盘中也会因为磁盘损坏而造成缓存丢失。一旦出现缓存丢失就需要重新执行整个计算过程，而执行整个计算过程就需要知道各个RDD之间的血缘关系，此时Spark会从Lineage记录中获取各个RDD间的血缘关系，然后重新计算，注意RDD是分区存放的，缓存丢失不可能所有分区的缓存数据都丢失，在恢复数据时只需要根据Lineage提供的血缘关系恢复相应丢失的分区即可。

3、RDD的CheckPoint技术

        在实际生产环境中，一个业务需求可能非常非常复杂，那么就可能调用很多算子，产生很多RDD，那么RDD之间的Lineage链条就会很长，一旦某个环节出现问题，比如上述缓存丢失后需要从头开始执行冗长的Lineage链条，容错的成本会很高。此时CheckPoint技术就出现了。

        CheckPoint技术也是将RDD的数据保存起来，但是它仅支持硬盘存储，并且它在设计上被认为是安全的，不保留血缘关系。如果使用了CheckPoint技术，记录血缘关系的Lineage就会被中断（严格来说CheckPoint之前的血缘关系就不记录了）。CheckPoint存储RDD数据，是集中收集各个分区的数据进行存储（存储到HDFS中，由于HDFS的安全性，因此CheckPoint被认为在设计上是安全的，在HDFS上存储的数据一般不会丢失，除非手动删除），而缓存是分散存储。

        使用者可以将重要的RDD CheckPoint下来，出错后，只需从最近的CheckPoint开始重新运算即可。类似与VMware虚拟机的快照，出错了可以切换到离当前最近的快照上重新执行。

        缓存与CheckPoint的对比：

        基于它们两者的特点，一般情况下， 数据量少，业务需求简单可以使用缓存技术（轻量化）；数据量大，业务需求复杂可以使用CheckPoint技术（重量级）。

        CheckPoint的实现： 

        打开hdfs的/output/ckp目录可以看到已经保存好的rdd3数据：

        打开4040端口，打开rdd6.collect所对应的子job对应的DAG图，发现job的运行是从CheckPoint开始的。

八、案例-搜索引擎日志分析案例

1、案例介绍

        我们使用由搜狗实验室提供的开源“用户查询日志数据（SogouQ.txt）”，使用Spark将数据封装到RDD中进行业务数据处理分析。数据在：http://www.sogou.com/labs/resource/q.php，请自行下载。数据格式如下：

2、业务需求

        A、搜索关键词统计，就是从用户搜索的内容中提取出关键词。比如用户搜索“我如果去清华大学学习能成才吗？”我们需要搜索出关键词：清华大学，学习，成才等等。这里需要使用python提供的第三方库jieba来实现中文分词。然后统计搜索关键词最多的5个。

        B、用户搜索点击统计，将用户ip与关键词拼接统计谁说的那个关键词最多

        C、搜索时间段统计，统计什么时候活跃用户比较多

3、jieba库的入门使用

        首先安装jieba库，pip install jieba（如果使用远程解析器需要在linux上安装jieba）

        返回的数据是无法直接查看的，是一个生成器类型。见结果强制类型转化为list：

4、功能实现

        需求一：

        点击运行main方法，我们会发现由于jieba.cut_for_search中文分词方法是机器基于一些算法自动进行的分词，分出来的词有些是不符合实际意义的，这里需要调整一下：

        在defs.py中增加新函数： 

        在主函数中添加：

        需求二：

        在defs.py中添加新函数

        此时就得知了各个用户的兴趣是什么，在后续营销过程中可以针对特点的用户推荐特定的产品，提高营销量。

        需求三：

        注意通过上述两个需求发现有某些函数经常会用到，例如lambda a，b：a + b，每次用都要自己写一下很不方便，其实python已经内置了对应的函数，导入直接用即可：

        from operator import add

5、将代码提交到集群中运行

        删除setMaster配置，将数据设置为hdfs中的数据，将代码上传到linux系统中，此外代码中使用了jieba库，保证每台服务器都安装了jieba

        在提交代码时可以使用额外的参数来使资源得到充分利用，在此之前需要查看一下集群的资源有多少：

        查看cpu核数：cat /proc/cpuinfo | grep processor | wc -l     # 查看processor有几个

        查看内存大小：free -g                      # 查看total

        在规划的时候要空闲出一部分供其他任务使用。

九、共享变量

1、广播变量

        上述代码在执行过程中本地list对象stu_info_list与分布式RDD对象有了联系，在执行过程中具体RDD计算之前的代码是Driver执行的，因此后续执行RDD计算需要将stu_info_list列表发送给相应的Task任务。假设当前的任务有多个Executor 进程，每个Executor进程内部有多个Task线程来执行任务。stu_info_list会分别发送给各个Task线程，我们知道线程之间是共享进程资源的，但是目前会将资源stu_info_list发送多份给Executor进程，造成了内存的浪费以及额外的网络IO。上述代码可以正常执行，只是有点浪费资源。

        解决方法：将本地list对象标记为广播变量，此时各个Executor只会收到一份数据集，内部各个线程（分区）之间共享这一份数据集。

        广播变量的实现：

         思考：为什么要使用本地集合对象，直接将其也设置为RDD分布式集合对象，这样两个对象都是在Executor中，这不就避免了网络传输？

        注意对于小文本数据，如果都设置为RDD对象，两个RDD对象之间关联需要用到rdd.join算子，执行过程中需要进行大量的shuffle操作，因此反而性能会下降。如果是大体量的数据就无法使用本地集合对象了，本地集合对象是有Driver执行的，会保存在Driver的内存中，如果数据量太大会导致Driver内存溢出。

2、累加器

        现在有一个需求：对map算子中的数据，进行计数累加，得到全部数据计算完后的累加结果

        由于设置了两个分区，因此计算过程是在各自分区内执行的。值得注意的是最后一个print的输出尽然是0，明明设置了global，按理说count的最终结果一定被修改了，为什么这里还是0呢？这就是分布式计算的累加问题。

        解决方法将count设置为累加器：

        累加器的实现：

        注意：累加器在使用过程中要注意RDD的数据是过程数据，RDD一旦执行完老的RDD就没了，基于这种特性在执行累加器时可能会出现与想法不一致的情况。 

        明明调用了一次map_func为什么结果为20，应该为10呀！！！。由于RDD是过程数据的特点，在执行完rdd2.collect()之后，rdd2已经被清理了，因此执行rdd3=rdd2.map（lambda x：x）时rdd2不存在，spark会按照血缘关系再次重新生成rdd2，相当于又执行了一次map_func，因此结果为20。

3、综合案例

        （1）数据准备：

        （2）需求

        对正常的单词进行单词计数

        对特殊字符出现的次数进行累加，特殊字符定义如下：

        abnormal_char = [",", ".", "!", "#", "$", "%"]

        （3）开发

        补充：在python中字符串str的方法：str.strip()