1. Spark核心概念
1.1 Spark简介
Apache Spark是新兴的一种快速通用的大规模数据处理引擎。它的优势有三个方面：
- 通用计算引擎 能够运行MapReduce、数据挖掘、图运算、流式计算、SQL等多种框架
- 基于内存 数据可缓存在内存中，特别适用于需要迭代多次运算的场景
- 与Hadoop集成 能够直接读写HDFS中的数据，并能运行在YARN之上
Spark是用Scala语言编写的，所提供的API也很好地利用了这门语言的特性，当然作为数据科学的一环，它也可以使用Java和Python编写应用。这里我们将用Python给大家做讲解。
1.2 Spark核心
Spark支持多种运行模式。单机部署下，既可以用本地（Local）模式运行，也可以使用伪分布式模式来运行；当以分布式集群部署的时候，可以根据实际情况选择Spark自带的独立（Standalone）运行模式、YARN运行模式或者Mesos模式。虽然模式多，但是Spark的运行架构基本由三部分组成，包括SparkContext（驱动程序）、ClusterManager（集群资源管理器）和Executor（任务执行进程）。
- SparkContext提交作业，向ClusterManager申请资源；
- ClusterManager会根据当前集群的资源使用情况，进行有条件的FIFO策略：先分配的应用程序尽可能多地获取资源，后分配的应用程序则在剩余资源中筛选，没有合适资源的应用程序只能等待其他应用程序释放资源；
- ClusterManager默认情况下会将应用程序分布在尽可能多的Worker上，这种分配算法有利于充分利用集群资源，适合内存使用多的场景，以便更好地做到数据处理的本地性；另一种则是分布在尽可能少的Worker上，这种适合CPU密集型且内存使用较少的场景；
- Excutor创建后与SparkContext保持通讯，SparkContext分配任务集给Excutor，Excutor按照一定的调度策略执行任务集。
Spark包含1个driver(笔记本电脑或者集群网关机器上)和若干个executor(在各个节点上)，通过SparkContext(简称sc)连接Spark集群、创建RDD、累加器（accumlator）、广播变量（broadcast variables），简单可以认为SparkContext（驱动程序）是Spark程序的根本。
Driver会把计算任务分成一系列小的task，然后送到executor执行。executor之间可以通信，在每个executor完成自己的task以后，所有的信息会被传回。
1.3 RDD(弹性分布式数据集)介绍
在Spark里，所有的处理和计算任务都会被组织成一系列Resilient Distributed Dataset(弹性分布式数据集，简称RDD)上的transformations(转换) 和 actions(动作)。
RDD是一个包含诸多元素、被划分到不同节点上进行并行处理的数据集合，可以将RDD持久化到内存中，这样就可以有效地在并行操作中复用（在机器学习这种需要反复迭代的任务中非常有效）。在节点发生错误时RDD也可以自动恢复。
说起来，RDD就像一个NumPy array或者一个Pandas Series，可以视作一个有序的item集合。
只不过这些item并不存在driver端的内存里，而是被分割成很多个partitions，每个partition的数据存在集群的executor的内存中。
1.4 RDD transformations和actions
大家还对python的list comprehension有印象吗，RDDs可以进行一系列的变换得到新的RDD，有点类似那个过程，我们先给大家提一下RDD上最最常用到的transformation:
map()对RDD的每一个item都执行同一个操作flatMap()对RDD中的item执行同一个操作以后得到一个list，然后以平铺的方式把这些list里所有的结果组成新的listfilter()筛选出来满足条件的itemdistinct()对RDD中的item去重sample()从RDD中的item中采样一部分出来，有放回或者无放回sortBy()对RDD中的item进行排序
特别注意：Spark的一个核心概念是惰性计算。当你把一个RDD转换成另一个的时候，这个转换不会立即生效执行！！！Spark会把它先记在心里，等到真的需要拿到转换结果的时候，才会重新组织你的transformations(因为可能有一连串的变换)
这样可以避免不必要的中间结果存储和通信。记住哦，transformation属于多行计算
刚才提到了惰性计算，那么什么东西能让它真的执行转换与运算呢？
是的，就是我们马上提到的Actions，下面是常见的action，当他们出现的时候，表明我们需要执行刚才定义的transform了:
-
collect(): 计算所有的items并返回所有的结果到driver端，接着collect()会以Python list的形式返回结果 -
first(): 和上面是类似的，不过只返回第1个item -
take(n): 类似，但是返回n个item -
count(): 计算RDD中item的个数 -
top(n): 返回头n个items，按照自然结果排序 -
reduce(): 对RDD中的items做聚合
1.5 针对更复杂的transformations和actions
咱们刚才已经见识到了Spark中最常见的transform和action，但是有时候我们会遇到更复杂的结构，比如非常非常经典的是以元组形式组织的k-v对（key, value）
我们把它叫做pair RDDs，而Sark中针对这种item结构的数据，定义了一些transformation和action:
-
reduceByKey(): 对所有有着相同key的items执行reduce操作 -
groupByKey(): 返回类似(key, listOfValues)元组的RDD，后面的value List 是同一个key下面的 -
sortByKey(): 按照key排序 -
countByKey(): 按照key去对item个数进行统计 -
collectAsMap(): 和collect有些类似，但是返回的是k-v的字典
2. PySpark之词频统计
首先，我们导入pyspark的包，创建SparkContext，建立RDD
import pyspark
from pyspark import SparkContext
from pyspark import SparkConf
conf=SparkConf().setAppName("miniProject").setMaster("local[*]")
sc=SparkContext.getOrCreate(conf)
然后读取文本文件
textFile = sc.textFile("file:///usr/local/spark/mycode/wordcount/word.txt")
textFile是一个方法，可以用来加载文本数据，默认是从HDFS上加载，如果要加载本地文件，就必须使用file:///加路径的形式
从文本中读取数据后就要开始进行词频统计了
wordCount = textFile.flatMap(lambda line:line.split(" ")).\
map(lambda word:(word, 1)).reduceByKey(lambda x,y:x+y)
flatMap会逐行遍历文本内容，然后对每行内容进行lambda函数的操作，即line:line.split(" ")，该操作会把每一行内容赋值给line，然后对每一个line进行split(" ")操作，即对每一行用空格分隔成单独的单词，这样每一行都是一个由单词组成的集合，因为有很多行，所以就有很多歌这样的单词集合，执行完 textFile.flatMap(lambda line:line.split(" "))后会把这些单词集合组成一个大的单词集合
map(lambda word:(word, 1))中的map对上述产生的单词集合进行遍历，对于每一个单词进行map函数内的操作，即lambda word:(word,1)，该操作会把每个单词赋值给word，然后组成一个键值对，这个键值对的key是这个单词，而value是1，这样就把每个单词变成了这个单词的键值对形式。执行完这个map后就会获得一个RDD，这个RDD的每一个元素是很多个键值对
reduceByKey(lambda x,y:x+y)会对RDD中的每个元素根据key进行分组，然后对该分组进行括号内的操作，即lambda x,y:x+y，通过对具有相同key的元素进行该操作，reduce操作会把具有相同key的元素的value进行相加，这样最后变成一个大的键值对，key是相同的，value是具有相同key的键值对的个数，这样，词频统计就完成了。
网友评论