Map-Reduce

我认为上图代表着MapReduce不仅仅包括Map和Reduce两个步骤这么简单，还有两个隐含步骤没有明确，全部步骤包括：切片、转换、聚合、叠加，按照实际的运算场景上述步骤可以简化。

具体的流程为：

• 原始数据 -) [切片] -> 数据对单元集合(列表） (k1,v1)
• 数据对单元集合 (k1,v1) -> [Map转换] -) 数据对单元集合 (k2,v2)
• 数据对单元集合 (k2,v2) -> [聚合(合并] -) 数据对单元集合(字典）(k2,[v2])
• 数据对单元集合 (k2,[v2]) -> [抽象(叠加)reduce] -) 数据对单元集合（列表）[(k3,v3)]

Spark作业流程

1. 用户编写Spark程序，定义RDD创建、转换、行动、存储等操作。提交程序（spark-submit xx.py)申请Spark集群执行该程序；
2. Spark集群收到用户的程序后，启动Driver进程，负责响应执行用户的函数。Driver也可以在Client端启动，也可以在某个worker节点启动； Driver启动后，发现函数包含了RDD的相关操作（这些操作会分散在集群中并行完成），与Master节点通信申请资源；
3. 在Spark集群中的所有worker节点都会向Master节点注册自己的计算资源；Master资源会通过心跳检测来检测节点状态；在收到driver在3条件下申请后，master会命令已注册的worker节点按照调度策略启动executor进程；
4. Master检查发现worker节点状态存活且worker节点上的executor启动成功，Master将已经启动的资源信息通知driver进程，让driver进程了解可以使用哪些资源来完成Spark程序；
5. driver根据Spark程序中的RDD操作情况对程序进行分割，分割后的任务发送给已经申请的多个executor资源，每个executor独立完成分配的计算任务，并将执行的结果反馈给driver，driver负责了解每一个executor进程的完成情况，以统一了解整个spark程序的完成情况；
6. worker节点上运行的executor是真正工作的执行者，在每个worker节点上可以启动多个executor，每个executor单独运行一个JVM进程，每个计算任务则是运行在executor中的一个线程，最后executor负责将计算结果保存到磁盘中；
7. driver通知client应用程序执行完成。
   
   上述的步骤中有三个细节需要特别注意：

• spark采用惰性计算来生成RDD（只有遇到action操作才会生成新的RDD），生成操作通过将所有相关操作形成一个有向无环图DAG来试试，每个DAG会触发Spark生成一个作业。
• driver节点中的DAGScheduler实例会想有向无环图中的节点依赖关系遍历，并将所有操作切分成多个调度阶段（stage)，并生成对应的taskSet。
• driver节点中的TaskScheduler会为每一个taskSet生成taskSetManager(管理每个taskset的内部调度任务，维护taskset的声明周期），并提交到worker节点的executor进程中执行。
• 从微观上来看，一个spark应用程序最终是分解成了多个taskset的任务急。在并行执行taskset的情况下，TaskScheduler调度策略可以分为FIFO先进先出和FAIR公平调度两种模式。

Spark并发机制

• Spark的基础数据结构RDD是由一系列的分区构成的，每个分区中包含一个RDD的部分数据；
• 在Spark调度和执行计算任务时，会为RDD的每个分区创建一个Task；
• 默认的情况下，Spark为每个需要执行的Task分配一个CPU内核资源
  
  
  
  

Spark RDD和Job生命周期

RDD 的生存周期主要包括 RDD 的创建和操作。RDD 的创建有两种方式：从 HDFS（或其他分布式存储系统）输入创建以及从父 RDD 转换得到新 RDD。对于 RDD 的操作可以分为两种，即 Transformation 和 Action。Transformation 是根据现有的数据集创建一个新的数据集，Action 是在 RDD 上运行计算后，返回一个值给驱动程序。Transformations 操作是Lazy 的，也就是说从一个 RDD 转换生成 转换生成另一个 RDD 的操作不马上执行，Spark 在遇到Transformations 操作时只会记录需要这样的操作，并不会去执行，需要等到有 Actions 操作的时候才会真正启动计算过程进行计算。

Job 的生存周期是从 Action 作用于某 RDD 时开始的，此时该 Action 将会作为一个 Job被提交。在提交的过程中，DAGScheduler 模块介入运算，计算 RDD 之间的依赖关系。RDD之间的依赖关系就形成了 DAG。每一个 Job 被分为多个 stage，划分 stage 的一个主要依据是当前计算因子的输入是否是确定的，如果是则将其分在同一个 stage，避免多个 stage 之间的消息传递开销。当 stage 被提交之后，由 TaskScheduler 根据 stage 来计算所需要的 task，并将 task 提交到对应的 Worker。

pyspark

• 数据集
• access.log:第一列为网络标号，0表示2G和3G的网络，1为4G网络。用以过滤和去重
• words.txt: 关于python的英文文档介绍，用于实现计数和topk。
• bayes1.dat: 朴素贝叶斯输入数据，有四个天气维度决定一个人是否出行，四个维度分别是天气外观、温度、湿度、刮风，最后一个值是否出行。
• bayes2.dat: 朴素贝叶斯测试数据。

[kejun.he@spark36 ~]$ cat access.log
0 12341234 www.baidu.com/s?wd=spark
1 22113345 www.hulian.com
1 332244 www.wanglian.com
0 15558 www.zhice.com
1 332244 www.wanglian.com
1 999222 www.up.com
1 999222 www.up.com
1 999222 www.up.com
[kejun.he@spark36 ~]$ cat bayes1.dat
sunny hot high FALSE no
sunny hot high TRUE no
overcast hot high FALSE yes
rainy mild high FALSE yes
rainy cool normal FALSE yes
rainy cool normal TRUE no
overcast cool normal TRUE yes
sunny mild high FALSE no
sunny cool normal FALSE yes
rainy mild normal FALSE yes
sunny mild normal TRUE yes
overcast mild high TRUE yes
overcast hot normal FALSE yes
rainy mild high TRUE no
[kejun.he@spark36 ~]$ cat bayes2.dat
sunny mild normal True
sunny hot high False
rainy cool high True 
overcast cool normal False
[kejun.he@spark36 ~]$ hadoop fs -put access.log /user/kejun.he/input
[kejun.he@spark36 ~]$ hadoop fs -put words.txt /user/kejun.he/input
[kejun.he@spark36 ~]$ hadoop fs -put bayes1.dat /user/kejun.he/input
[kejun.he@spark36 ~]$ hadoop fs -put bayes2.dat /user/kejun.he/input

过滤，去重及统计

在access.log中统计4G网络的不同用户访问次数。

>>> log=sc.textFile("file:///home/kejun.he/access.log")
>>> log.collect()
[u'0 12341234 www.baidu.com/s?wd=spark', u'1 22113345 www.hulian.com', u'1 332244 www.wanglian.com', u'0 15558 www.zhice.com', u'1 999221 www.up.com/s?blabla', u'1 332214 www.wanglian.com', u'1 999221 www.up.com', u'1 999222 www.up.com', u'1 999223 www.up.com']
>>> filter_rdd=log.filter(lambda x:x.split(" ")[0]=="1") #导入数据
>>> filter_rdd.collect()
[u'1 22113345 www.hulian.com', u'1 332244 www.wanglian.com', u'1 999221 www.up.com/s?blabla', u'1 332214 www.wanglian.com', u'1 999221 www.up.com', u'1 999222 www.up.com', u'1 999223 www.up.com']
>>> import re
>>> map_rdd=filter_rdd.map(lambda x:" ".join([x.split(" ")[1],re.compile("/.*").sub("",x.split(" ")[2])]))#数据转换：1去除网络标记号,2.保留站点信息
>>> map_rdd.collect()
[u'22113345 www.hulian.com', u'332244 www.wanglian.com', u'999221 www.up.com', u'332214 www.wanglian.com', u'999221 www.up.com', u'999222 www.up.com', u'999223 www.up.com']
>>> revert_rdd=map_rdd.map(lambda x:(x.split(" ")[1],x.split(" ")[0])) #数据转换，形成<网站，用户>的key-value对
>>> revert_rdd.collect()
[(u'www.hulian.com', u'22113345'), (u'www.wanglian.com', u'332244'), (u'www.up.com', u'999221'), (u'www.wanglian.com', u'332214'), (u'www.up.com', u'999221'), (u'www.up.com', u'999222'), (u'www.up.com', u'999223')]
>>> distinct_rdd=revert_rdd.distinct() #去重
>>> distinct_rdd.collect()
[(u'www.hulian.com', u'22113345'), (u'www.up.com', u'999222'), (u'www.wanglian.com', u'332214'), (u'www.up.com', u'999221'), (u'www.up.com', u'999223'), (u'www.wanglian.com', u'332244')]
>>> set_one_rdd=distinct_rdd.mapValues(lambda x:1) #去除用户信息
>>> set_one_rdd.collect()
[(u'www.hulian.com', 1), (u'www.up.com', 1), (u'www.wanglian.com', 1), (u'www.up.com', 1), (u'www.up.com', 1), (u'www.wanglian.com', 1)]
>>> from operator import add
>>> reduce_rdd=set_one_rdd.reduceByKey(add)#统计次数
>>> reduce_rdd.collect()
[(u'www.up.com', 3), (u'www.hulian.com', 1), (u'www.wanglian.com', 2)]
>>> reduce_rdd.saveAsTextFile("output") #保存统计结果

相关计数

与之前的Spark环境安装 实现的简单计数有区别，相关计数是计算两个或多个实体以一定方式共同出现的次数或者概率，在数据挖掘的领域中，我们也称为co-occurance。这个工作将切换到python文件运行的模式，不在pyspark下使用。

运算的步骤如下：

• 获取每一行的文本数据
• 对该行数据进行文本清洗
• 清洗后进行转换关联，并形成key-value对
• reduce统计

from pyspark import SparkContext
from operator import add
import re

print ("start doing ###############")
sc=SparkContext("spark://spark36.localdomain:7077","myApp")
naive_words=sc.textFile("hdfs:///user/kejun.he/input/words.txt").cache()
words_rdd=naive_words.map(lambda str1: str1.replace("."," ").replace(","," ").replace("("," ").replace(")"," ").replace("-"," ").replace(":"," ").split(" ")).cache()
#print words_rdd.collect()

def cut_words(x):
    l=len(x)
    i=0
    while(i<l-1):
        if is_valid(x[i]) and is_valid(x[i+1]):
            yield ((x[i],x[i+1]),1)
        i+=1

def is_valid(ch):
    match=re.search('^[a-zA-Z1-9]+$', ch)
    if match:
        return True
    else:
        return False

transform_rdd=words_rdd.flatMap(cut_words)
#print transform_rdd.collect()

count_result=transform_rdd.reduceByKey(add)

sorted_result=count_result.sortBy(lambda x:x[1])
#print (sorted_result.collect())

sorted_result.repartition(1).saveAsTextFile("count_co_occurance")

朴素贝叶斯