Spark快速大数据分析

Spark

第1章 Spark数据分析导论

1.1 Spark是什么

实现快速而通用的集群计算的平台

1.2 一个大一统的软件栈

Spark core

实现了Spark的基本功能，包括任务调度、内存管理、错误恢复、与存储系统交互，对弹性分布式数据集（resilient distribution dataset，RDD）的API定义。RDD表示分布在多个计算节点上可以并行操作的元素集合。是Spark的主要编程抽象。

Spark Sql

Spark操作结构化数据的程序包。支持多种数据源，如Hive表、Parquet以及JSON等。

Spark Streaming

对实时数据进行流式计算的组件。

MLlib

机器学习包，分类、回归、聚类、协同过滤

GraphX

操作图的程序库，可以进行并行的图计算

集群管理器

支持在各种集群管理器上工作

1.3 Spark的用户与用途

数据科学任务

分析数据，交互性Shell

数据处理应用

为开发用于集群并行执行的程序提供了一条捷径

1.4 Spark简史

第2章 Spark下载与入门

2.1 下载Spark

2.2 Spark中Python和Scala的Shell

2.3 Spark核心概念简介

每个Spark应用由一个驱动程序（driver program）来发起集群上的各种并行操作。驱动器程序通过一个SparkContext对象来访问Spark。这个对象代表计算集群的一个连接。Shell启动时已经自动创建一个sc变量。驱动器程序一般要管理多个执行器（executor）节点。

2.4 独立应用

Python中需要使用 bin/spark-submit my_script.py 执行，才会对spark相关组件进行加载

from pyspark import SparkConf, SparkContext
conf = SparkConf().setMaster("local").setAppName("My App")
sc = SparkContext(conf= conf)

第3章 RDD编程

RDD其实就是分布式的元素集合，在Spark中，对数据的所有操作不外乎创建RDD、转换已有RDD以及调用RDD操作进行求值。

3.1 RDD基础

RDD支持两种类型操作

• 转换操作（transformation）
  会将一个RDD生成一个新的RDD
• 行动操作（action）
  会对RDD计算出一个结果，并将结果返回到驱动程序中，或把结果存储到外部存储系统中
  Spark会惰性计算RDD，只有第一次在一个行动操作中用到才会真正计算，而且由于大量的转换连接在一起，其可以一次计算，避免中间结果的保存
  cache()与使用默认级别的persist()是一样的

3.2 创建RDD

• 使用sc.parallelize(x)将已有的集合创建RDD
• 使用sc.textFile(f)从外部文件读取并创建RDD

3.3 RDD操作

• 转换操作是返回新的RDD
  许多转换操作都是针对各个元素的，不过并不是所有的操作都是这样的
  filter、map、
  转换操作可以操作任意个的输入RDD，A.union(B)
  Spark会使用谱系图（lineage graph）来记录这些不同RDD之间的依赖关系，Spark使用这些信息来按需计算每个RDD，在持久化的RDD丢失部分数据时恢复也会使用
• 行动操作，会强制执行那些求值必须用到的RDD的转换操作
  count()\first()\take(x)\collect()\saveAsTextFile()
• 惰性求值
  不应该将RDD看着放着特定数据的数据集，而最好把每个RDD当作我们通过转换操作创建出来的、记录如何计算数据的指令列表，把数据读取到RDD的操作同样是惰性的。在Spark中，写一个非常复杂的映射不一定比使用很多简单的连续操作获得好得多的性能，用户可以用更小的操作来组织它们的程序，使操作更好管理。

3.4 向Spark传递函数

• Python
  使用lambda表达式或者顶层函数或者定义的局部函数。Python会在你不经意间把函数所在的对象也序列化传递出去。

rdd.filter(lambda x: a.name in x) # 整个a对象被序列化并传递
rdd.filter(lambda x: name in x) # 传递name

• Scale
• Java

3.5 常见的转换操作和行动操作

• 基本RDD
  
  • 针对各个元素的转换操作
    
    • map(lambda_)
    • filter(lambda_)
    • flatMap(lambda_) 每个元素生成多个元素，返回的不是一个元素，而是一个返回值序列的迭代器。输出的RDD是包含各个迭代器可访问的所有元素的RDD。
  • 伪集合操作
    
    • distinct() 生成唯一值，开销很大，需要将数据通过网络进行混洗（shuffle）
    • union(other_rdd) 该函数会包含重复数据如果有的话
    • intersection(other_rdd) 会去掉所有重复的数据，包括单个数据集中的，需要网络混洗
    • subtract(other_rdd) 返回所有只存在于rdd1而不存在于rdd2中的，需要混洗，如果rdd1中重复且不在rdd2中的则保留
    • cartesian(other) 计算笛卡尔积，形成(Ai, Bj)对
    • sample(withReplacement, fraction, [seed]) 对RDD进行采样，以及是否替换
  • 行动操作
    
    • count()
    • countByValue() 各元素在rdd中出现的个数
    • take(n) 返回n个元素，并且只访问尽量少的分区，会得到不均衡的集合
    • top(num) 从rdd中返回最前面的num个元素
    • takeOrdered(num) 从rdd按照顺序返回最前面的num个
    • takeSample(withReplacement, num, seed) 获得一个采样
    • reduce(lambda_)
    • fold(zero, fun) 两者都要求返回值类型和rdd中元素类型操作，fold需要提供初始值
    • aggregate(intiv, fun1, fun2) 提供类型初值，函数1对元素合并起来放入累加器，函数2对累加器进行两两合并，可返回与输入类型不同的类型
    • collect() 会将整个rdd的内容返回，通常在测试中使用
    • foreach(fun) 对每个元素进行操作，而不需要将rdd发回本地
• 在不同的RDD类型间转换
  
  • mean、variance等函数都实现在基本的RDD类中

get_max_min = x.aggregate((-inf,inf), 
     (lambda acc, value:(max(acc[0],value),min(acc[1],value))),
     (lambda acc1,acc2:(max(acc1[0],acc2[0]),min(acc1[1],acc2[1]))))
x.takeOrdered(10,key=lambda x:-x)
x.fold(0, (lambda acc,x: x+acc))

3.6 持久化（缓存）

默认的缓存级别，scale和java会将数据以序列化的形式缓存在JVM的堆空间中。在Python中，我们会始终序列化要持久化存储的数据，所以持久化级别默认值就是以序列化后的对象存储在JVM堆空间中。

# rdd.persisit()/rdd.cache()
rdd.count()
rdd.collect().mkstring(',') ## 会导致两次rdd计算，除非前面添加 

可以通过unpersist来取消缓存

第4章 键值对操作

4.1 动机

为包含键值对类型的RDD提供了一些专有的操作。这些RDD被称为pair RDD。PairRDD提供了并行操作各个键或者跨节点重新进行数据分组的操作接口。

4.2 创建Pair RDD

Python中需要返回一个二元组组成的RDD。只能是元组，字典不是，底层操作应该是采用迭代器协议遍历的。

x = sc.parallelize([(x,y) for x in 'abcde' for y in range(6,10)])

4.3 pair RDD的转换操作

针对单个pair RDD的操作

聚合操作

• reduceByKey(fun) 合并具有相同键的值
• combineByKey(createCombiner, mergeValue, mergeCombiners, partitioner) 使用不同类型合并具有相同键的值
  createCombiner 每个分区第一次遇见该key时创建
  mergeValue 合并combiner和key的新数据
  mergeCombiners 合并不同分区的同键combiner
  partitioner
• mapValues(fun) 对每个值应用函数而不改变键
• flatMapValues(fun) 对每个值应用一个返回迭代器的函数，然后对返回的每个元素都生成一个对应原键的键值对记录，通常用于符号化
• keys() 返回一个仅包含键的RDD
• values() 返回一个仅包含值的RDD
• sortByKey() 返回一个根据键排队的RDD

x = sc.parallelize([(1,2),(3,4),(3,6)])
x.reduceByKey(lambda x,y: x+y).collect()  #[(1, 2), (3, 10)]
x.foldByKey(2, lambda acc,x: x+acc).collect()  #[(1, 4), (3, 14)]
x.mapValues(lambda x:x+1).collect()  #[(1, 3), (3, 5), (3, 7)]
# 求键的平均
x.groupByKey().mapValues(lambda x:(sum(x), len(x))).collect() #[(1, (2, 1)), (3, (10, 2))]
x.mapValues(lambda x:(x,1)).reduceByKey(lambda x,y:(x[0]+y[0], x[1]+y[1])).collect()
x.combineByKey(
    (lambda x: (x,1)),
    (lambda acc, x:(acc[0]+x, acc[1]+1)),
    (lambda acc,acc2:(acc[0]+acc2[0], acc[1]+acc2[1]))
).map(lambda x: (x[0],x[1][0]/x[1][1])).collectAsMap()

数据分组

• groupByKey(fun) 对具有相同键的值进行分组 结果为[k,iterable(v)]
  rdd.groupByKey().mapValues(lambda value:value.reduce(fun))与rdd.groupByKey(fun)等价，后者更快速
• groupBy(fun) 使用fun作用于元素上返回的键值进行分组 结果为[k,iterable(v)]

连接

• join(other) 对两个RDD进行内连接，类似表格中的连接
• rightOuterJoin(other) 不存在为None
• leftOuterJoin(other)
• cogroup(other) 将两个rdd中拥有相同键的数据分组到一起，返回结果为[(k,(iterable(V),iterable(W)))...]，可以用来求交集，可以用于三个以及以上的RDD

排序

• sortByKey(ascending, keyfunc=(lambda x:len(x)))

行动操作

• countByKey() 返回[(k,count)...]
• collectAsMap() 返回{k:v, ...}
• lookup(key) 返回指定键对应的所有值 返回[v1,v2...]

其他

• subtractByKey(other) 删除与other中键相同的元素

每个RDD都有固定数目的分区，分区数决定了在RDD上执行操作时的并行度。大多数操作符都支持接受第二个参数，该参数用来指定分组结果或聚合结果的RDD的分区数

r.reduceByKey((lambda x,y: x+y), 10)
r.getNumPartitions() #查看当前分区数
r.coalesce(num)
r.repartition(num)

4.5 数据分区

如果给定RDD只需要被扫描一次，完全没有必要对其预先进行分区处理。只有当数据集多次在诸如连接这种基于键的操作中使用时，分区才会有帮助。

Spark没有给出显式控制每个键具体落在哪一个节点上的方法（原因是节点可能失败），但其确保同一组的键出现在同一个节点上。

例如Join操作，如果有个用户信息表，不变化，而产生新的交易记录需要对表进行join，如果不控制，则可能对两者都进行混洗后再连接，如果将信息表先Hash分区并持久化后，则不再需要对信息表进行混洗，只需要对交易数据进行混洗

user_data = user_data_old.partitionBy(num).persist()

获取RDD分区的方式

python中未提供API

从分区中获益的操作

类似cogroup/groupwith/join/leftouterJoin/rightOuterJoin/groupByKey/reduceByKey/combineByKey/lookup

影响分区方式的操作

Spark知道各操作会如何影响分区方式，并将会对数据进行分区的操作的结果RDD自动设置为对应的分区器。例如join结果会自动按照哈希分区，后续reduceByKey操作则会变快。

转换操作不一定会保留分区方式，例如map是可以改变键的。Spark提供了mapValues/flatMapValues可以保持键值不变

会设置分区方式的操作：cogroup/groupwith/join/left../right../groupByKey/reduceByKey/combineByKey/partitionBy/sort
取决于父RDD的分区方式：mapValues/flatMapValues/filter
二元操作符：取决于父RDD的分区方式，默认情况下会采用hash，数目与并行度一致。如何一个父亲设置过分区方式，则以它为准，如果两者都设过，则以第一个为准

实例：pagerank

pages = [
    ['a',['b','c','d','e']],
    ['b',['a','c','e']],
    ['c',['a','d','e']],
    ['d',['a']],
    ['e',['c','d']]
]
links = sc.parallelize(pages)
ranks = links.mapValues(lambda x:1.0)
for i in range(20):
    contributes = links.join(ranks).flatMap(lambda x:((i, x[1][1]/len(x[1][0])) for i in x[1][0])).reduceByKey(lambda x,y:x+y)
    ranks = contributes.mapValues(lambda x:0.15+0.85*x)
ranks.collect()

自定义分区方式

第5章

第6章 Spark编程进阶

6.1 简介

两种类型的共享变量：

• accumulator 累加器：用来对信息进行聚合
• broadcast variable 广播变量：用来高效分发较大的对象

6.2 累加器

使用map或者filter传条件的时候，可以使用驱动器程序中定义的变量，但是集群中的每个任务都会得到这些变量的一份新的副本。更新这些副本的值不会影响到驱动器中的对应变量

比如说统计空行

blanklines = sc.accumulator(0)
def extr(line):
    global blanklines
    if (line == ''):
        blanklines += 1
    return line.split(' ')
result = file.flatMap(extr)

可以通过map和reduce进行聚合，但是这种操作更方便

• 通过在驱动器中调用sc.accumulator(initvalue)创建，返回值为sc.Accumulator[T]类型的对象
• Spark闭包中可以使用 += 增加累加器的值（Java：add）
• 驱动程序可以调用累加器的value属性访问累计器，可以读取和设置
• 工作节点上的任务不能访问累加器的值，累加器是一个只写变量

Spark有时候任务会失败或者会有较慢机器，因此可以会多次计算累计器

• 对于行动操作中的累加器，Spark只会把每个任务对累加器的修改应用一次，因此需要可靠则需要将其放到foreach中
• 对于rdd转换操作中使用的累加器，就不能保证了，比如rdd被移除缓存后，又重新计算时，其累加器会再次更新
• 可以自定义累加器

6.3 广播变量

Spark会自动将闭包中所有引用到的变量发送到工作节点上。方便但是低效：1）默认的任务发射机制是专门为小任务进行优化的；2）事实上在多个并行操作中使用同一个变量，则在第二次操作会再发送一次该变量，广播变量的类型为spark.broadcast.Broadcast[T]，可以对该变量调用value属性来获取实际的对象的值，只会发送到节点一次。

• 通过对一个类型T的对象调用sc.broadcast()创建出一个spark.broadcast.Broadcast[T]对象，任何可序列化的类型都可以这么实现
• 通过value属性访问该变量的值
• 变量只会发送到各节点一次，应作为只读处理（修改该值不会影响到其他节点）
• 可以选择spark.serialize或其他序列化方法进行自定义

6.4 基于分区进行操作