Spark入门阶段一之扫盲笔记

摘要：同时集成了机器学习类库。基于计算框架，将的分布式计算应用到机器学习领域。提供了一个简单的声明方法指定机器学习任务，并且动态地选择最优的学习算法。宣称其性能是的多倍。

spark是分布式并行数据处理框架
与mapreduce的区别：
mapreduce通常将中间结果放在hdfs上，spark是基于内存并行大数据框架，中间结果放在内存，对于迭代数据spark效率更高，mapreduce总是消耗大量时间排序，而有些场景不需要排序，spark可以避免不必要的排序所带来的开销，spark是一张有向无环图，spark支持scala，python，java等
适用范围：
spark更适合于迭代云端比较多的ml和dm运算，因为spark里面有rdd的抽象概念，spark比hadoop更通用，spark提供的数据集操作类型有很多，不像hadoop只提供map和reduce俩种操作，比如map，filter，flatmapt，sample，groupbykey，reducebykey，union，join，cogroup，mapvalues，sort，partionby等多种操作类型，spark
把这些操作称为transformations，同时还提供count，collect，reduce，lookup，save等多种action操作。这些多种多样的数据集操作类型，给开发上层应用的用户提供了方便，各个处理节点之间的通信模型不在像hadoop那样就是唯一的data shuffle一种模式，用户可以明明，物化，控制中间结果的存储，分区等，可以说编程模型比hadoop更灵活。

spark是基于内存的迭代计算框架，使用与需要多次操作特定数据集的应用场合，需要反复操作的次数越多，所需要读取的数据量越大，受益越大，数据量小但是计算密集度较大的场合，受益就相对较小. 不过由于rdd的特性，spark不适用那种一部细粒度更新状态的应用，例如web服务的存储或者增量的web爬虫和索引，就是对于那种增量修改的应用模型不合适。

spark和hadoop的结合：
spark可以直接对hdfs进行数据的读写，同样支持spark on yarn。spark可以与mapreduce运行于同集群中，共享存储资源与计算，数据仓库shark实现上借用hive，几乎和hive完全兼容。

四种spark运行模式，local模型用于测试开发，standlone 独立集群模式，spark on yarn spark在yarn上 ，spark on mesos spark在mesos上。

应用：
企业大数据应用： 1，count 平均值 2.分类，对比 3.趋势，统计分析 4，精准预测 人工智能
行业大数据案例：电商，传媒，能源，交通

spark生态系统介绍：
spark 可以很容易和yarn结合，直接调用HDFS、Hbase上面的数据，和hadoop结合。
spark核心部分分为RDD。Spark SQL、Spark Streaming、MLlib、GraphX、Spark R等核心组件解决了很多的大数据问题

Spark分为driver和executor，driver提交作业，executor是application早worknode上的进程，运行task，driver对应为sparkcontext。Spark的RDD操作有transformation、action。Transformation对RDD进行依赖包装，RDD所对应的依赖都进行DAG的构建并保存，在worknode挂掉之后除了通过备份恢复还可以通过元数据对其保存的依赖再计算一次得到。当作业提交也就是调用runJob时，spark会根据RDD构建DAG图，提交给DAGScheduler，这个DAGScheduler是在SparkContext创建时一同初始化的，他会对作业进行调度处理。当依赖图构建好以后，从action开始进行解析，每一个操作作为一个task，每遇到shuffle就切割成为一个taskSet，并把数据输出到磁盘，如果不是shuffle数据还在内存中存储。就这样再往前推进，直到没有算子，然后运行从前面开始，如果没有action的算子在这里不会执行，直到遇到action为止才开始运行，这就形成了spark的懒加载，taskset提交给TaskSheduler生成TaskSetManager并且提交给Executor运行，运行结束后反馈给DAGScheduler完成一个taskSet，之后再提交下一个，当TaskSet运行失败时就返回DAGScheduler并重新再次创建。一个job里面可能有多个TaskSet，一个application可能包含多个job。

1、shark介绍：
shark基本上就是spark的框架基础上提供和hive一样的hivesql命令接口，为了最大程度的保持和hive的兼容性，shark使用hive的api来实现query parsing和logic plan generation，最后的physicalplan execution阶段用spark代替hadoop mapreduce，用过配置shark参数，shark可以自动在内存中缓存特定的rdd，实现数据重用，进而加快特定数据集的检索，同时，shark通过udf用户自定义函数实现特定的数据分析学习算法，使得sql数据查询和运算分析能结合在一起，最大化rdd的重复使用。

2、spark streaming介绍：
Spark Streaming 是 Spark 提供的对实时数据进行流式计算的组件，一般与kafka结合，基本的原理是将stream数据分成小的时间片段，以类似batch批量处理的方式来处理这些小部分数据。spark streaming构建在spark上，一方面是因为spark的低延迟执行引擎可以用于实时计算，此外小批量的处理方式使得他可以同时兼容批量和实时数据处理的逻辑和算法，方便了一些需要历史数据和实时数据联合分析的特定应用场景。
Spark Streaming也有一个StreamingContext，其核心是DStream，是通过以组时间序列上的连续RDD来组成的，包含一个有Time作为key、RDD作为value的结构体，每一个RDD都包含特定时间间隔的数据流，可以通过persist将其持久化。在接受不断的数据流后，在blockGenerator中维护一个队列，将流数据放到队列中，等处理时间间隔到来后将其中的所有数据合并成为一个RDD(这一间隔中的数据)。其作业提交和spark相似，只不过在提交时拿到DStream内部的RDD并产生Job提交，RDD在action触发之后，将job提交给jobManager中的JobQueue，又jobScheduler调度，JobScheduler将job提交到spark的job调度器，然后将job转换成为大量的任务分发给spark集群执行。

3、Graphx
主要用于图的计算。核心算法有PageRank、SVD奇异矩阵、TriangleConut等。

4、Spark SQL
是Spark新推出的交互式大数据SQL技术。把sql语句翻译成Spark上的RDD操作可以支持Hive、Json等类型的数据。

5、Spark R
通过R语言调用spark，目前不会拥有像Scala或者java那样广泛的API，Spark通过RDD类提供Spark API，并且允许用户使用R交互式方式在集群中运行任务。同时集成了MLlib机器学习类库。

6、MLBase
从上到下包括了MLOptimizer（给使用者）、MLI（给算法使用者）、MLlib（给算法开发者）、Spark。也可以直接使用MLlib。ML Optimizer，一个优化机器学习选择更合适的算法和相关参数的模块，还有MLI进行特征抽取和高级ML编程 抽象算法实现API平台，MLlib分布式机器学习库，可以不断扩充算法。MLRuntime基于spark计算框架，将Spark的分布式计算应用到机器学习领域。MLBase提供了一个简单的声明方法指定机器学习任务，并且动态地选择最优的学习算法。

7、Tachyon
高容错的分布式文件系统。宣称其性能是HDFS的3000多倍。有类似java的接口，也实现了HDFS接口，所以Spark和MR程序不需要任何的修改就可以运行。目前支持HDFS、S3等。

rdd是spark最基本，也是最根本的数据抽象，RDD表示分布在多个计算节点上的可以并行操作的元素集合，rdd是只读的，分区记录的集合。
rdd支持两种操作，1，转换从现有的数据集创建一个新的数据集，2，动作 在数据集上运行计算后，返回一个值给驱动程序，例如，map就是一种转换，他将数据集每一个元素都传递给函数，并返回一个新的分布数据集表示结果，另一个方面，reduce是一个动作，通过一些函数将所有的元组叠加起来，并将结果返回给driver程序，spark中的所有转换都有惰性的，也就是说，他们并不会直接计算结果，相反的，他们只是记住应用哦个到基础数据集上的这些转换动作，例如，我们可以实现，通过map创建的一个新数据集，并在reduce使用，最终只返回reduce的结果给driver，而不是整个大的新数据集。默认情况下，每个转换过的rdd都会在你在他之上执行一个动作时被重新计算，不过，你也可以使用persist方法，持久话一个rdd在内存中，在这种情况下，spark将会在集群中，保存相关元素，下次你查询这个rdd是，他将能更快访问，在磁盘上持久化数据集，或在集群间赋值数据集也是支持的。除了这些操作外，用户还可以请求将rdd缓存起来，而且，用户还可以通过partitioner类获取rdd的分区顺序，然后将另一个rdd按照同样的方式分区。

如何操作rdd？
1、如何获取rdd 1，从共享的文件系统获取，hdfs，2.通过已存在的rdd转换 3.将已存在的scala集合并行化，通过调用sparkcontext的parallelize方法实现 4.改变现有rdd的之久性，rdd是懒散，短暂的
2、操作rdd的俩个动作，1，actions：对数据集计算后返回一个数值value给驱动程序，例如redue将数据集的所有元素用某个函数聚合后，将最终结果返回给程序，2.transformation 根据数据集创建一个新的数据集，计算后返回一个新rdd；例如map将数据的每个元素讲过某个函数计算后，返回一个姓的分布式数据集。

actions具体内容：

reduce（func）通过函数func聚集数据集中所有元素，func函数接受2个参数，返回一个值，这个函数必须是关联性的，确保可以被正确的并发执行。

collect（） 在driver的程序中，以数组的形式，返回数据集的所有元素，这通常会在使用filter或者其他操作后，返回一个纵沟小的数据自己在使用，直接将整个rdd集coloect返回，很可能会让driver程序oom。

count（） 返回数据集的元素个数

take（n） 返回一个数组，用数据集的前n个元素组成，注意，这个操作目前并非在多个节点上，并行执行，而是driver程序所在机制，单机计算所有的元素：注；gateway的内存压力会增大，需要谨慎使用

first（）返回数据集的第一个元素

saveAsTextFile(path) 将数据集的元素，以txtfile的形式，保存到本地文件系统，hdfs或者其他hadoop支持的文件系统，spark将会调用每个元素的tostring方法，并将他转换成文件中一行文本。

saveAsSequenceFile（path）将数据集的元素，以sequencefile的格式，到指定的目录下，本地系统，hdfs或者其他hadoop支持的文件系统，rdd的元组必须有key-value对组成，并都实现了hadoop的- writable接口或隐式可以转换为wirtable

foreach（func）在数据集的每个元素上，运行函数func，这通常用于更新一个累加器变量，或者和外部存储系统做交互。直接使用 rdd.foreach(println) 在local模式下是可行的，但是在cluster模式下是不行的，必须要执行collect()方法，将所有的数据拉取到本地，然后执行foreach()操作。如果是数据量比较小的话可以使用take方法，rdd.take(100).foreach(println)

transformation具体内容：

map（func） 返回一个新的分布式数据集，有每个原元素经过func函数转换后组成

filter（func） 返回一个新的数据集，有经过func函数后返回值为true的原元素组成

flatmap（func）类似于map 但是每一个输入元素，会被映射0到多个输出元素，因此func函数的返回值是一个seq，而不是单一元素

sample（withReplacement，frac，seed） 给定的随机种子seed，随机抽样出数量为frac的数据

union（otherdataset）返回一个新的数据集，由原数据集和参数联合而成

intersection ： 只返回两个RDD中都有的元素，intersecton（）在运行时会去掉所有重复的元素（单个RDD内重复元素也会一起移除）。 需要通过网络混洗来发现共有数据。

distinct ： 生成一个只包含不同元素的新RDD。需要注意：distinct（） 操作的开销很大，因为它需要将所有数据通过网络进行混洗（shuffle），以确保每个元素只有一份。

subtract ： 接受另一个RDD作为参数，返回一个由只存在在第一个RDD而不存在第二个RDD中的所有元素组成的RDD。 需要数据混洗。

cartesian ： 返回所有可能的（a，b）对，其中a是源RDD中的元素，b是另一个RDD中的元素。

groupbykey（【num tasks】）在一个有kv对组成的数据集上调用，返回一个k，seq【v】对的数据集，注意，默认情况下，使用8个并行任务进行分组，你可以传入num task可选参数，根绝数据量设置不同数目的task

reducebykey（func，【num tasks】）在一个kv对的数据集上使用，返回一个kv的数据集，key相同的值都被使用指定的reduce函数聚合在一起，和groupbykey类似，任务个数是第二个参数来配置

join（otherdataset，【num tasks】）在类型kev和kw类型的数据集上调用，返回一个k（v w）对，每个key中所有元素都在一起的数据集

groupwith（otherdataset，【num tasks】）在类型为kv和kw类型的数据集上调用，返回一个数据集，组成元组为k seq【v】seq[w]tuples ，这个在其他框架称为cogroup

cartesian（otherdataset） 笛卡儿积，但在数据集t和u调用是，返回一个tu对的数据集，所有元素交互进行笛卡儿积。

持久化（缓存）

persist（）

cache()

每个saprk应用都有一个驱动器程序来发起集群上的各种并行操作。驱动器程序通过一个SparkContext对象来访问Spark。这个对象代表对计算集群的一个连接。
一旦有了SparkContext，你就可以用它来创建RDD。要执行这些操作，启动器程序一般要管理多个执行器（executor）节点。
可以先通过SparkConf对象来配置你的应用，然后基于这个SparkConf创建一个SparkContext对象。
创建SparkConf的基本方法，传递两个参数：
1、集群URL：告诉Spark如何连接到集群上。
2、应用名：当连接到一个集群式，这个值可以帮助你在集群管理器的用户界面中找到你的应用。

关闭Spark：调用SparkContext的stop（）方法。或直接退出应用。（system.exit（0）/sys.exit（））
在Spark中，对数据的所有操作不外乎是： 创建RDD、 转化已有的RDD、调用RDD操作进行求值
Spark中的RDD是一个不可变的分布式对象集合。每个RDD都被分为多个分区，这些分区运行在集群中的不同节点上。
当我们调用一个新的行动操作时，整个RDD都会从头开始计算。要避免这种行为，用户可以将中间结果持久化。

1、初始化sparkcontext

from pyspark import SparkConf, SparkContxt
conf = SparkConf().setMaster("local").setAppName("my app")
sc = SparkContext(conf=conf)

# 关闭连接
sc.stop()

2、RDD编程

# 从文件读取数据
line = sc.textFile("README.md")
# parallelize 方法
line = sc.parallelize(["pandas","i like pandas"])


inputRDD = sc.textFile("log.txt")
errRDD = inputRDD.filter(lambda x:"error" in x)
warnRDD = inputRDD.filter(lambda x:"warning" in x)
bindRDD = errRDD.union(warnRDD)


bindRDD.count()
bindRDD.take(10)
# 返回全部数据集
bindRDD.collect()


# lambda 函数
word = rdd.filter(lambda s:"python" in s)
# def 定义的函数
def containsErr(s):
    return "error" in s
word = rdd.filter(containsErr)

2.1、RDD常见转换操作
以 rdd={1,2,3,3} 为例的转换操作

# 将函数应用与RDD中的每个元素，将返回值构建新的RDD
rdd.map(x => x+1)

# 将函数应用用RDD中的每个元素，将返回的迭代器的所有内容构成新的RDD。通常用于切分单词。
rdd.flatMap(x=>x.to(3))  --> {1,2,3,2,3,3,3})

# 返回一个由通过传给filter()的函数的元素组成的RDD
rdd.filter(x=>x!=1)  -->  {2,3,4}

# 去重
rdd.distinct()  -->  {1,2,3}

sample(withReplacement,fraction,[seed])
# 对RDD进行采样，以及是否替换
rdd.sample(false,0.5)   -->  非确定的

以{1,2,3}和{3,4,5}的RDD转换操作

# 求并集
rdd.union(other)  --> {1,2,3,4,5}

# 求交集
rdd.intersection(other)  --> {3}

# 移除一个RDD中的内容，相当于减去一个交集
rdd.subtract(other)  -->  {1,2}

# 与另一个RDD的笛卡尔积
rdd.cartesian(other)  --> {(1,3),(1,4)...(3,5)}

2.2、RDD常见行动操作
以{1,2,3,3}为列说明常见行动操作

# 返回RDD中所有的元素
rdd.collect()  --> {1,2,3,3}

# 计数
rdd.count()

# 各元素在RDD中出现的次数
rdd.countByValue()   --> {(1,1),(2,1),(3,2)}

take(num)
# 返回前n元素

top(n)
# 排序后的前n个元素

# 按照指定顺序，从rdd中返回前n个元素
rdd.takeOrdered(2)(myOrdering)   --> {3,3}

takeSample(withReplacement,num,[seed])
# 从RDD中返回任意一些元素
rdd.takeSample(false,1)  --> 非确定的

# 并行整合rdd中所有的数据，比如sum
rdd.reduce((x,y)=>x+y)  --> 9

fold(zero)(func)
# 和reduce()一样，但是需要提供初始值
rdd.fold(0)((x,y)=>x+y) --> 9

aggregate(zeroValue)(seqOp,combOp)
# 和reduce类似，但是通常返回不同类型的函数
aggregate((0,0))((x,y)=>(x._1+y,x._2+1),
                 (x,y)=>(x._1+y._1,x._2+y._2) )  --> 9

# 对RDD中的每个元素使用给定的函数
rdd.foreach(func)

2.3、持久化缓存

from pyspark.storage import StorageLvel
rdd.presist(StoragLevel.DISK_ONLY)

RDD.cache()

# 缓存的级别
# MEMORY_ONLY
# MEMORY_ONLY_SER
# MEMORY_AND_DISK  # 如果内存放不下，则溢出写到磁盘上
# MEMORY_AND_DISK_SER  # 如果内存放不下，则溢出写到磁盘上，在内存中存放序列化后的数据
# DISK_ONLY

# 移除缓存
RDD.unpersist()

3、键值对操作

# 以{(1,2),(3,4),(3,6)}为例

# 合并具有形同键的值
rdd.reduceByKey((x,y)=>x+y)  -->{(1,2),(3,10)}

# 对具有相同键的值分组
rdd.groupByKey()  --->{(1,[2]),(3,[4,6])}

combineByKey(createCombiner,mergeValue,mergeComBiners,partitioner)
# 使用不同的返回类型合并具有相同键的值。有多个参数分别对应聚合操作的各个阶段，因而非常适合用来解释聚合操作各个阶段的功能划分。
# 下面是求每个键的平均值
sumCount=num.combineByKey((lambda x:(x,1)),
                           (lambda x,y:(x[0]+y,x[1]+1)),
                           (lambda x,y:(x[0+y[0],x[1]+y[1]])))
sumCount.map(lambda key,xy:(key,xy[0]/xy[1])).collectAaMap()

# 对pairRDD的每个值应用一个函数而不改变键
rdd.mapVlues(x=>x+1)

# 对pairRDD的每个值应用一个返回迭代器的函数，然后对返回的每个元素都生成一个对应原键的键值对记录，通常用于符号化
rdd.flatMapValues(x=>(x to 5))  -->{(1,2),(1,3),(1,4),(1,5),(3,4),(3,5)}

# 返回一个仅含有键的RDD
rdd.keys()   ->{1,3,3}

# 返回一个仅包含值的RDD
rdd.values()   -->{2,4,6}

# 返回一个根据键排序的RDD
rdd.sortByKey(ascending=True)  -->{(1,2),(3,4),(3,6)}

3.1、两个键值对RDD的转换操作

# 以rdd={(1,2),(3,4),(3,6)} other={(3,9)} 为例

# 删除rdd中键与other中键相同的元素
rdd.subtracByKey(other)  --> {(1,2)}

# 对两个rdd内链接
rdd.join(other)   --> {(3,(4,9)),(3,(6,9))}

# 对两个rdd进行连接操作，确保第一个rdd中的键必须存在（右外链接）
rdd.rightOuterJoin(other)   --> {(3,(some(4),9)),(3,(some(6),9))}

# 对两个rdd进行连接操作，确保第二个rdd中的键必须存在（左外连接）
rdd.leftOuterJoin(other)  --> {(1,(2,None)),(3,(4,some(9))),(3,(6,some(9)))}

# 将两个rdd中拥有相同键的数据分组到一起
rdd.congroup(other)  --> {(1,([2],[])),(3,([4,6],[9]))}

3.2、键值对Pair RDD的行动操作

# 以 rdd={(1,2),(3,4),(3,6)} 为例

# 对每个键对应的元素分别计数
rdd.countByKey()  --> {(1,1,),(3,2)})

# 将结果以映射表的形式返回，以便查询
rdd.collectAsMap()  --> Map{(1,2),(2,6)}

# 返回给定键对应的所有值
rdd.lookup(3)   --> [4,6]

4、并行度调优
每个rdd都有固定数目的分区，分区数决定了在rdd上执行操作的并行度。 大多数操作符都能接受第二个参数，用来指定分组结果或者聚合结果的rdd的分区数。
比如 sc.parallelize(data).reduceByKey(lambda x,y:x+y,10) 指定分区数10
查看分区数 rdd.partitions.size或rdd.getNumPartitions ，改变分区的方法repartition()

5、数据读取与保存
读取txt文件，输入的每一行都会成为RDD的一个元素。

# 读取文件
input=sc.textFile("file:///home/holden/README.md")
# 保存文件
result.saveAsTextFile(outputFile)

读取json

# 将json文件的每一行假设为一条记录来处理
import json
data = input.map(lambda x:json.load(x))
# 写
(data.filter(lambda x:x[lovesPandas"]).map(lambda x:json.dumps(x)).saveAsTextFile(outputFile))

读取csv,同样是将读取的文本的每一行当做一条记录

import csv
from io import StringIO
def loadRecord(line):
    """解析一行csv记录"""
    input = StringIO(line)
    reader = csv.DictReader(input,filednames=["name","favouriteAnimal"])
    return reader.next()
input = sc.textFile(inputFile).map(loadRecord)

# 保存csv
def writeRecords(records):
    """写出一些csv记录"""
    output = StringIO()
    writer = csv.DictWriter(output,fieldnames=["name","favoriteAnimal"])
    for record in records:
        writer.writerow(record)
    return [output.getvalue()]
pandaLovers.mapPartitions(writeRecords).saveAsTextFile(outputFile)

读取SequenceFile
Hadoop输入输出格式
关系型数据库
HBase

6、Spark进阶编程
6.1、两种类型的共享变量

累加器（qccumulator）：用于对信息聚合，提供了将工作节点中的值聚合到驱动器程序中的简单语法。

广播变量（broadcast variable）：用来高效分发较大的对象，让程序高效地向所有工作节点发送一个较大的值，以供一个或多个spark操作使用。

# 在python中累加空行,使用了累加器
file = sc.textFile(inputFile)
# 创建累加器并初始化为0
blankLine=sc.accumulator(0)

def extractCallSigs(line):
    global blankLine  # 访问全局变量
    if (line==""):
        blankLine+=1
    return line.split(" ")

callSigns = file.flatMap(extractCallSigns)
callSigns.saveAsTextFile(output)


# 使用广播变量查询国家
# 查询rdd中呼叫号对应的位置，将呼号前缀读取为国家代码来进行查询
signPrefixes = sc.broadcast(loadCallSignTable())   # 广播变量

def processSignCount(sign_count,signPrefixes):
    country=lookupCountry(sign_count[0],signPrefixes.value)
    count = sign_count[1]
    return (country,count)

countryContactCounts=(contactCounts.map(processSignCount).reduceByKey((lambda x,y:x+y)))
countryContactCounts.saveAsTextFile(output)

基于分区进行操作
spark提供基于分区的map和foreach，使部分代码只对rdd的每个分区运行一次，可以帮助降低这些操作的代价。

# 按照分区执行的操作符

mapPartitions()
# 参数：该分区中元素的迭代器。返回：元素的迭代器
# 对于RRD[T]的函数签名 ：f:(iterator[T])  --> iterator[U]

mapPartitionsWithIndex()
# 参数：分区序号，以及每个分区中的元素的迭代器。返回：元素的迭代器
# 对于RRD[T]的函数签名 ：f:(int,iterator[T])  --> iterator[U]

foreachPartitions()
# 参数：元素迭代器。返回：无
# 对于RRD[T]的函数签名 ：f:(iterator(T))  -->Unit

数值RDD的操作

count()
# RDD中元素个数
mean()
# 元素平均值
sum()
# 
max()
min()
variance()  # 方差
sampleVariance()  # 从采样中计算出的方差
stdev()  # 标准差
sampleStdev()   # 采用的标准差

7、基于MLlib的机器学习

# 逻辑回归的垃圾邮件分类
from pyspark.mllib.regression import LabeldPoint
from pyspark.mllib.feature import HashingTF
from pyspark.mllib.classification import LogisticRegressionWithSGD

spam=sc.textFile("spam.txt")
normal = sc.textFile("normal.txt")

# 创建一个HashingTF实例来把邮件文本映射为包含10000个特征的向量
tf=HashingTF(numFeatures=10000)
# 各邮件都切分为单词，每个单词映射为一个特征
spamFeatures = spam.map(lambda email: tf.transForm(email.split(" ")))
normalFeatures = normal.map(lambda email: tf.transform(email.split(" ")))

# 创建LabelPoint数据集分别存放阳性（垃圾邮件）和阴性（正常邮件）的例子
positiveExample = spamFeatures.map(lambda features:LabeldPoint(1,features))
negativeExamples = normalFeatures.map(lambda features:labeldPoint(0,features))
trainingData = positiveExample.union(negativeExample)
trainingData.cache()  # 因为逻辑回归是迭代算法，所以需要缓存训练数据RDD

# 使用SGD算法
model = LogisticRegressionWithSGD.train(trainningData)

# 以阳性和阴性的例子分别测试。
# 首先用一样的HashingTF特征来得到特征向量，然后对该向量应用得到的模型
posTest = tf.transform("O M G GET cheap stuff by sending money to ...".split(" "))
negTest = tf.transform("Hi Dad, i started studying spark the other ...".split(" "))
print( "predict for postive test example:%g" % model.predict(posTest))
print( "predict for negative test example:%g" % model.predict(negTest))

MLlib包含一些特有的数据类型，对于Scala和Java，它们位于org.apache.spark.mllib下，对于Python则是位于pyspark.mllib下。

spark有两个重要的抽象：

RDD，分布式弹性数据集，他是一个跨越多个节点的分布式集合。

另一个抽象是共享变量。spark支持两种类型的共享变量：一个是广播（broadcast variables）他可以缓存一个值在集群的各个节点。另一个是累加器（accumulators）他只能执行累加的操作，比如可以做计数器和求和。

初始化 Spark
在一个Spark程序中要做的第一件事就是创建一个SparkContext对象来告诉Spark如何连接一个集群。为了创建SparkContext，你首先需要创建一个SparkConf对象，这个对象会包含你的应用的一些相关信息。这个通常是通过下面的构造器来实现的：
new SparkContext(master, appName, [sparkHome], [jars])
参数说明：

master：用于指定所连接的 Spark 或者 Mesos 集群的 URL。

appName ：应用的名称，将会在集群的 Web 监控 UI 中显示。

sparkHome：可选，你的集群机器上 Spark 的安装路径（所有机器上路径必须一致）。

jars：可选，在本地机器上的 JAR 文件列表，其中包括你应用的代码以及任何的依赖，Spark 将会把他们部署到所有的集群结点上。
在 python 中初始化，示例代码如下：

//conf = SparkContext("local", "Hello Spark")
conf = SparkConf().setAppName("Hello Spark").setMaster("local")
sc = SparkContext(conf=conf)

说明：如果部署到集群，在分布式模式下运行，最后两个参数是必须的，第一个参数可以是以下任一种形式：
Master URL 含义

local 默认值，使用一个 Worker 线程本地化运行(完全不并行)

local[N] 使用 N 个 Worker 线程本地化运行，N 为 * 时，表示使用系统中所有核

local[N,M] 第一个代表的是用到的核个数；第二个参数代表的是容许该作业失败M次

spark://HOST:PORT 连接到指定的 Spark 单机版集群 master 进程所在的主机和端口

mesos://HOST:PORT 连接到指定的 Mesos 集群。host 参数是Moses master的hostname。端口默认是5050
如果你在一个集群上运行 spark-shell，则 master 参数默认为 local。在实际使用中，当你在集群中运行你的程序，你一般不会把 master 参数写死在代码中，而是通过用 spark-submit 运行程序来获得这个参数。但是，在本地测试以及单元测试时，你仍需要自行传入 local 来运行Spark程序。

运行代码有几种方式，一是通过 spark-shell 来运行 scala 代码，一是编写 java 代码并打成包以 spark on yarn 方式运行，还有一种是通过 PySpark 来运行 python 代码。

在 spark-shell 和 PySpark 命令行中，一个特殊的集成在解释器里的 SparkContext 变量已经建立好了，变量名叫做 sc，创建你自己的 SparkContext 不会起作用。

 
    org.apache.spark
    spark-core_2.10
    2.1.1
  
  
    junit
    junit
    4.12
    test
  

创建一个简单的spark程序：

public class SimpleApp {
  public static void main(String[] args) {
      // 文件路径
      String logFile = "/home/wm/apps/spark-1.4.0-bin-hadoop2.6/README.md";
      SparkConf conf = new SparkConf().setAppName("Simple Application").setMaster("local");
      JavaSparkContext sc = new JavaSparkContext(conf);
      JavaRDD logData = sc.textFile(logFile).cache();
      @SuppressWarnings("serial")
      long numAs = logData.filter(new Function() {
          public Boolean call(String s) throws Exception {
              return s.contains("a");
          }

      }).count();
      @SuppressWarnings("serial")
      long numBs = logData.filter(new Function() {

          public Boolean call(String s) throws Exception {
              return s.contains("b");
          }

      }).count();
      System.out.println("Lines with a: " + numAs + ", lines with b: " + numBs);
      sc.close();
  }
}

Spark的核心就是围绕着RDD，它是一个自动容错的分布式数据集合。他有两种方式创建，第一种就是在驱动程序中对一个集合进行并行化。第二种是来源于一个外部的存储系统。比如：共享系统、HDFS、HBase或者任何提供任何Hadoop 输入格式的数据源。

第一种：Parallelized Collections 创建这个集合需要调用那个JavaSparkContext的parallelize方法来初始化一个已经存在的集合。

List data = Arrays.asList(1,2,3,4,5);
JavaRDD distData = sc.parallelize(data);

这就创建了一个并行的集合，在这个集合上可以执行 distData.reduce((a, b) -> a + b)
在并行数组中一个很重要的参数是partitions，它来描述数组被切割的数据集数量。Spark会在每一个partitions上运行任务，这个partitions会被spark自动设置，一般都是集群中每个CPU上运行2-4partitions，但是也可以自己设置，可以通过parallelize （e.g. sc.parallelize(data, 10)），在有些地方把partitions成为 slices。

第二种：External Datasets

JavaRDD distFile = sc.textFile("data.txt");

textFile也可以设置partitions参数，一般都是一个block一个partitions，但是也可以自己设置，自己设置必须要不能少于block的数量。
针对Hadoop的其他输入格式，你能用这个JavaSparkContext.hadoopRDD方法，你需要设置JobConf和输入格式的类。也可以使用JavaSparkContext.newAPIHadoopRDD针对输入格式是基于“new”的MapReduceAPI

先将测试数据上传到 hdfs:
$ hadoop fs -put access.log
然后，编写一个 python 文件，保存为 SimpleApp.py：

from pyspark import SparkContext

logFile = "access.log"

sc = SparkContext("local", "Simple App")

rdd = sc.textFile(logFile).cache()

counts = rdd.map(lambda line: line.split()[8]).map(lambda word: (word, 1)).reduceByKey(lambda a, b: a + b).sortByKey(lambda x: x) 

# This is just a demo on how to bring all the sorted data back to a single node.  
# In reality, we wouldn"t want to collect all the data to the driver node.
output = counts.collect()  
for (word, count) in output:  
    print "%s: %i" % (word, count)  

counts.saveAsTextFile("/data/result")

sc.stop()

接下来，运行下面代码：

$ spark-submit  --master local[4]   SimpleApp.py

JavaRDD lines = sc.textFile("data.txt");
JavaPairRDD pairs = lines.mapToPair(s -> new Tuple2(s, 1));
JavaPairRDD counts = pairs.reduceByKey((a, b) -> a + b);

 
    org.apache.spark
    spark-core_2.10
    2.11


    mysql
    mysql-connector-java
    5.1.13


    org.apache.hadoop
    hadoop-client
    2.6.0


    junit
    junit
    4.12
    test

由于需要读取HDFS中的数据，所以需要hadoop-client文件

在main函数中首先创建JavaSparkcontext对象。

SparkConf conf = new SparkConf().setAppName("FindError");
JavaSparkContext sc = new JavaSparkContext(conf);

/**
* 
* 列出指定目录中的文件，这里的文件是不包括子目录的。
* @param pathOfDirectory
*     目录路径
* @return
* @throws IOException 
*/
public static String[] findFilePathFromDir(String dst) throws IOException {
  Set filePathSet = new HashSet();
  String[] result = null;
  Configuration conf = new Configuration();
  FileSystem fs = FileSystem.get(URI.create(dst), conf);
  FileStatus fileList[] = fs.listStatus(new Path(dst));
  int size = fileList.length;
  for (int i = 0; i < size; i++) {
      filePathSet.add(fileList[i].getPath().toString());
  }
  if (filePathSet.size() > 0) {
      result = new String[filePathSet.size()];
      int i = 0;
      for (String str : filePathSet) {
          result[i++] = str;
      }
  }
  fs.close();
  return result;
}

依次遍历文件路径并为每个文件创建一个新的RDD然后计算出这个文件中包涵ERROR字符串的行数。

Map result = new HashMap();
if (filePaths != null) {
  for (String path : filePaths) {
      result.put(path, sc.textFile(path).filter(new Function() {

          public Boolean call(String line) throws Exception {
              return line.contains("ERROR");
          }

      }).count());
  }
}

将results中的数据写入mysql中

/**
* 将结果写入mysql中
* @param result
* @throws Exception 
*/
public static void wirteResultToMysql(Map result) throws Exception {
  String DBDRIVER = "com.mysql.jdbc.Driver";  
  //连接地址是由各个数据库生产商多带带提供的，所以需要多带带记住  
  String DBURL = "jdbc:mysql://ip:3306/test";  
  //连接数据库的用户名  
  String DBUSER = "root";  
  //连接数据库的密码  
  String DBPASS = "root";
  Connection con = null; //表示数据库的连接对象  
  PreparedStatement pstmt = null; //表示数据库更新操作  
  String sql = "insert into aaa values(?,?)";  
  Class.forName(DBDRIVER); //1、使用CLASS 类加载驱动程序  
  con = DriverManager.getConnection(DBURL,DBUSER,DBPASS); //2、连接数据库  
  pstmt = con.prepareStatement(sql); //使用预处理的方式创建对象  
  if (result != null) {
      for (String str : result.keySet()) {
          pstmt.setString(1, str);
          pstmt.setLong(2, result.get(str));
          pstmt.addBatch();
      }
  }
  //pstmt.executeUpdate(); //执行SQL 语句，更新数据库  
  pstmt.executeBatch();
  pstmt.close();  
  con.close(); // 4、关闭数据库  
}

通常情况下，当一个函数传递给一个在远程集群节点上运行的Spark操作（比如map和reduce）时，Spark会对涉及到的变量的所有副本执行这个函数。这些变量会被复制到每个机器上，而且这个过程不会被反馈给驱动程序。通常情况下，在任务之间读写共享变量是很低效的。但是，Spark仍然提供了有限的两种共享变量类型用于常见的使用场景：广播变量和累加器。
1、广播变量
广播变量允许程序员在每台机器上保持一个只读变量的缓存而不是将一个变量的拷贝传递给各个任务。它们可以被使用，比如，给每一个节点传递一份大输入数据集的拷贝是很低效的。Spark 试图使用高效的广播算法来分布广播变量，以此来降低通信花销。 可以通过 SparkContext.broadcast(v) 来从变量 v 创建一个广播变量。这个广播变量是 v 的一个包装，同时它的值可以功过调用 value 方法来获得。以下的代码展示了这一点：

broadcastVar = sc.broadcast([1, 2, 3])


>>> broadcastVar.value
[1, 2, 3]

在广播变量被创建之后，在所有函数中都应当使用它来代替原来的变量v，这样就可以保证v在节点之间只被传递一次。另外，v变量在被广播之后不应该再被修改了，这样可以确保每一个节点上储存的广播变量的一致性（如果这个变量后来又被传输给一个新的节点）。

2、累加器
累加器是在一个相关过程中只能被”累加”的变量，对这个变量的操作可以有效地被并行化。它们可以被用于实现计数器（就像在MapReduce过程中）或求和运算。Spark原生支持对数字类型的累加器，程序员也可以为其他新的类型添加支持。累加器被以一个名字创建之后，会在Spark的UI中显示出来。这有助于了解计算的累进过程（注意：目前Python中不支持这个特性）。

可以通过SparkContext.accumulator(v)来从变量v创建一个累加器。在集群中运行的任务随后可以使用add方法或+=操作符（在Scala和Python中）来向这个累加器中累加值。但是，他们不能读取累加器中的值。只有驱动程序可以读取累加器中的值，通过累加器的value方法。

以下的代码展示了向一个累加器中累加数组元素的过程：

accum = sc.accumulator(0)
Accumulator

>>> sc.parallelize([1, 2, 3, 4]).foreach(lambda x: accum.add(x))
...
10/09/29 18:41:08 INFO SparkContext: Tasks finished in 0.317106 s

scala> accum.value
10

这段代码利用了累加器对 int 类型的内建支持，程序员可以通过继承 AccumulatorParam 类来创建自己想要的类型支持。AccumulatorParam 的接口提供了两个方法：zero用于为你的数据类型提供零值；addInPlace 用于计算两个值得和。比如，假设我们有一个 Vector类表示数学中的向量，我们可以这样写：

class VectorAccumulatorParam(AccumulatorParam):
    def zero(self, initialValue):
        return Vector.zeros(initialValue.size)

    def addInPlace(self, v1, v2):
        v1 += v2
        return v1

# Then, create an Accumulator of this type:
vecAccum = sc.accumulator(Vector(...), VectorAccumulatorParam())

累加器的更新操作只会被运行一次，Spark 提供了保证，每个任务中对累加器的更新操作都只会被运行一次。比如，重启一个任务不会再次更新累加器。在转化过程中，用户应该留意每个任务的更新操作在任务或作业重新运算时是否被执行了超过一次。

累加器不会改变Spark 的惰性求值模型。如果累加器在对RDD的操作中被更新了，它们的值只会在启动操作中作为 RDD 计算过程中的一部分被更新。所以，在一个懒惰的转化操作中调用累加器的更新，并没法保证会被及时运行。 下面的代码段展示了这一点：

accum = sc.accumulator(0)
data.map(lambda x => acc.add(x); f(x))
// 这里，accum任然是0，因为没有action算子，所以map也不会进行实际的计算

参考官方文档：http://spark.apache.org/docs/...
spark-submit
首先需要打包代码，如果你的代码需要依赖其他的包环境则需要多带带的打包这些依赖，应为cluster会将所有依赖的jar包分发到各个节点上进行使用。推荐的方法是将依赖包和程序都统一的打成一个包，这样就可以直接使用spark-submit方法来运行，具体的pom.xml配置如下：

     
        org.apache.spark
        spark-core_2.10
        2.11
        provided
    
    
        mysql
        mysql-connector-java
        5.1.13
    
    
        org.apache.hadoop
        hadoop-client
        2.6.0
        provided
    
    
        junit
        junit
        4.11
        test
    


    
        
            org.apache.maven.plugins
            maven-compiler-plugin
            2.3.2
            
                
                1.7
                1.7
            
        
        
            maven-assembly-plugin
            2.5.5
            
                
                    jar-with-dependencies
                
            
            
                
                    make-assembly
                    package
                    
                        single
                    
                
            
        
    

spark && hadoop 的scope值都设置为provided
在服务器上提交的命令如下：

./bin/spark-submit 
  --class 
  --master  
  --deploy-mode  
  --conf = 
  ... # other options
   
  [application-arguments]

spark-submit 可以加载一个配置文件，默认是加载在conf/spark-defaults.conf

Spark对所有常见的单元测试框架提供友好的支持。你只需要在测试中创建一个SparkContext对象，然后吧master URL设为local，运行测试操作，最后调用 SparkContext.stop() 来停止测试。注意，一定要在 finally 代码块或者单元测试框架的 tearDown方法里调用SparkContext.stop()，因为Spark不支持同一程序中有多个SparkContext对象同时运行。

1、Spark Standalone Mode
除了运行在Mesos和YARN集群之外，spark也提供了简单的独立部署模式。可以通过手动的启动master和worker，也可以通过spark提供的启动脚本来启动。独立部署也可以通过运行在一个机器上，进行测试。
为了安装你需要放置一个编译好的spark版本到每个机器上。
启动集群有两种方式，一种是手动启动，另一种是通过启动脚本启动。
1.1、手动启动spark集群
启动一个独立的master可以使用如下的命令：
./sbin/start-master.sh
一旦启动可以通过访问：http://localhost:8080端口访问master
可以使用如下的命令来使worker节点连接到master上：
./sbin/start-slave.sh
worker在加入到master后可以访问master的http://localhost:8080，可以看到被加入的worker节点的信息。
在启动master和worker的时候可以带上参数进行设置，参数的列表如下:其中比较重要的是：
-c CORES， 这个是指定多少个cpu分配给spark使用，默认是全部cpu
-m MEM，这个是指定多少的内存分配给spark使用，默认是全部的内存的减去1g的操作系统内存全部分配给spark使用。一般的格式是1000M or 2G
-d DIR, 这个指定spark任务的日志输出目录。
–properties-file FILE 指定spark指定加载的配置文件的路径默认是： conf/spark-defaults.conf

1.2、脚本方式部署
通过spark的部署脚本部署首先需要在spark的主目录下创建一个conf/slaves的文件，这个文件中每一行代表一个worker的hostname.需要注意的是，master访问worker节点是通过SSH访问的，所以需要master通过ssh无密码的登录到worker，否则需要设置一个 SPARK_SSH_FOREGROUND的环境变量，这个变量的值就是每个worker的密码

然后可以通过spark安装目录下的sbin/….sh文件进行启动， 如果要启动和停止master和slave可以使用：
sbin/start-all.sh
sbin/stop-all.sh
注意的是这些脚本必须是在master机器上执行

同时你可以通过配置集群的 conf/spark-env.sh文件来进一步配置集群的环境。但是这也文件需要通过拷贝conf/spark-env.sh.template文件来创建，并且需要把这个文件拷贝到所有的worker节点上。
其中： SPARK_MASTER_OPTS && SPARK_WORKER_OPTS 两个配置项比较复杂。
通过在SparkContext构造器中传入spark://IP:PORT这个来启用这个集群。同时可以在交互式的方式启动脚本中使用：./bin/spark-shell –master spark://IP:PORT 来启动集群执行。
独立部署模式的集群现在只是简单的支持FIFO调度。 为了允许多个并发用户，可以通过SparkConf设置每个应用程序需要的资源的最大数。默认情况下，它会请求使用集群的全部的核，而这只是同时运行一个应用程序才回有意义。

val conf = new SparkConf()
             .setMaster(...)
             .setAppName(...)
             .set("spark.cores.max", "10")
val sc = new SparkContext(conf)

除了可以在程序中指定你也可以在spark-env.sh中设置默认的值，export SPARK_MASTER_OPTS="-Dspark.deploy.defaultCores="

2、spark的高可用设置
spark的高可用设置有两种，一种是通过Zookeeper来实现，另一种是通过本地文件系统来实现。

2.1、使用ZooKeeper备份master，利用zookeeper提供的领导选举和状态保存，你可以让更多的master连接到zookeepre实例。一个将会被选举为leader其他的则会保存备份他的状态。如果master死掉，zookeeper可以选举一个新的leader，整个过程需要1到2分钟的时间，但是这个过程只会对新的任务调度有影响。为了使用这种方式需要的配置项为：SPARK_DAEMON_JAVA_OPTS，这个配置项有三个配置信息：spark.deploy.recoveryMode/spark.deploy.zookeeper.url/spark.deploy.zookeeper.dir

2.2、使用本地文件系统来恢复该节点。为了使用这种方式需要的配置项为：SPARK_DAEMON_JAVA_OPTS，这个配置项有两个配置信息：spark.deploy.recoveryMode、spark.deploy.recoveryDirectory

Spark架构采用了分布式计算中的Master-Slave模型。Master是对应集群中的含有Master进程的节点，Slave是集群中含有Worker进程的节点。Master作为整个集群的控制器，负责整个集群的正常运行；Worker相当于是计算节点，接收主节点命令与进行状态汇报；Executor负责任务的执行；Cluster作为用户的客户端负责提交应用，Driver负责控制一个应用的执行。
Spark集群部署后，需要在主节点和从节点分别启动Master进程和Woker进程，对整个集群进行控制。在一个Spark应用的执行过程中，Driver和Worker是两个重要角色。Driver程序是应用逻辑执行的起点，负责作业的调度，即Task任务的分发，而多个Worker用来管理计算节点和创建Executor并行处理任务。在执行阶段，Driver会将Task和Task所依赖的
file和jar序列化后传递给对应的Worker机器，同时Exucutor对相应数据分区的任务进行处理。
下面详细介绍Spark的架构中的基本组件。

ClusterManager：在Standalone模式中即为Master（主节点），控制整个集群，监控Worker。在YARN模式中为资源管理器。

Worker：从节点，负责控制计算节点，启动Executor或Driver。在YARN模式中为NodeManager，负责计算节点的控制。
Spark整体流程为：Client提交应用，Master找到一个Worker启动Driver，Driver向Master或者资源管理器申请资源，之后将应用转化为RDD Graph，再由DAGScheduler将RDD Graph转化为Stage的有向无环图提交给TaskScheduler，由TaskScheduler提交任务给Executor执行。在任务执行过程中，其他组件协同工作，确保整个应用顺利进行。

Application：应用。可以认为是多次批量计算组合起来的过程，在物理上可以表现为你写的程序包+部署配置。应用的概念类似于计算机中的程序，它只是一个蓝本，尚没有运行起来。

RDD：Resilient Distributed Datasets，弹性分布式数据集。RDD即是计算模型里的一个概念，也是你编程时用到的一种类。一个RDD可以认为是spark在执行分布式计算时的 一批相同来源、相同结构、相同用途的数据集，这个数据集可能被切割成多个分区，分布在不同的机器上，无论如何，这个数据集被称为一个RDD。在编程 时，RDD对象就对应了这个数据集，并且RDD对象被当作一个数据操作的基本单位。比如，对某个RDD对象进行map操作，其实就相当于将数据集中的每个 分区的每一条数据进行了map映射。

Partition：分区。一个RDD在物理上被切割成多个数据子集，分布在不同的机器上。每个数据子集叫一个分区。

RDD Graph：RDD组成的DAG（有向无环图）。RDD是不可变的，一个RDD经过某种操作后，会生成一个新的RDD。这样说来，一个 Application中的程序，其内容基本上都是对各种RDD的操作，从源RDD，经过各种计算，产生中间RDD，最后生成你想要的RDD并输出。这个 过程中的各个RDD，会构成一个有向无环图。

Lineage：血统。RDD这个概念本身包含了这种信息“由哪个父类RDD经过哪种操作得到”。所以某个RDD可以通过不断寻找父类，找到最原始的那个RDD。这条继承路径就认为是RDD的血统。

Job：从Application和RDD Graph的概念可以知道，一个应用往往对应了一个RDD Graph。这个应用在准备被spark集群运行前，实际上就是会生成一个或多个RDD Graph结构，而一个RDD Graph，又可以生成一个或多个Job。一个Job可以认为就是会最终输出一个结果RDD（后面会介绍，实际上这是action操作）的一条由RDD组 织而成的计算，在Application生成的RDD Graph上表现为一个子图。Job在spark里应用里也是一个被调度的单位。

宽依赖：RDD生成另一个RDD时，各个两个父子RDD间分区的对应关系，被叫做RDD间依赖。宽依赖就是子RDD的某个分区，依赖父RDD的全部分区。

窄依赖：窄依赖就是子RDD的某个分区，只依赖常数个父RDD的分区。宽窄依赖的区别如下图所示。

Stage：Stage可以理解为完成一个Job的不同阶段。一个Job被划分为多个Stage，每个Stage又包含了对多个RDD的多个操作。一个Stage里，一般包含了一个宽依赖操作，或者多个窄依赖操作。
窄依赖是指前一个rdd计算能出一个唯一的rdd，比如map或者filter等；宽依赖则是指多个rdd生成一个或者多个rdd的操作，比如groupbykey reducebykey等，这种宽依赖通常会进行shuffle。

算子：父子RDD间的某种操作，被叫某种算子。比如下面会介绍的map，filter，groupByKey等。算子可从多个维度分类，之后再介绍。

Task：一个分区对应一个Task。实际上一个Task就是在一个Stage范围内，某个Executor所要执行的算子。

TaskSet：一个Stage范围内，所有相同的Task被称为一个TaskSet。

DAGScheduler：DAGScheduler用于根据RDD DAG切分Stage，并维护各个Stage的先后依赖关系，相当于完成了一个Job内的不同Stage间的调度策略。

TasksetManager：管理一个TaskSet，并决定了这个TaskSet中各个Task的分发策略。

TaskScheduler：执行实际的Task分发操作。

SparkUI: 4044
History Server:18080
怎么看？http://www.cnblogs.com/xing90...

http://blog.csdn.net/qq_26562...
http://blog.csdn.net/suzyu123...
http://www.cnblogs.com/helloc...
http://blog.csdn.net/suzyu123...
http://www.jianshu.com/nb/340...
http://www.cnblogs.com/ainima...
http://www.chinahadoop.cn/gro...
https://yq.aliyun.com/article...
http://ifeve.com/category/spa...