1.Action2.Transformation3.key-valueRDD的操作4.缓存操作5. 共享变量
from pyspark import SparkConf,SparkContext
import findspark
findspark.init()
conf=SparkConf().setAppName('RD').setMaster('local[*]')
sc=SparkContext(conf=conf)
1.Action
| collect | 将数据汇集到Driver,数据过大时有超内存风险 |
|---|---|
| take | 将前若干个数据汇集到Driver，相比collect安全 |
| takeSample | 可以随机取若干个到Driver,第一个参数设置是否放回抽样 |
| first、count | 取第一个数和统计数量 |
| reduce | 利用二元函数对数据进行规约 rdd.reduce(lambda x,y:x+y) |
| foreach | accum = sc.accumulator(0) rdd.foreach(lambda x:accum.add(x)) accum.value |
| countByKey | 按key统计个数 |
| saveAsTextFile | 保存rdd成text文件到本地 重新读入会被解析文本 rdd_loaded = sc.textFile(text_file) |
| map | 对每个元素进行映射 rdd.map(lambda x:x**2).collect() |
|---|---|
| filter | 根据条件过滤数据 rdd.filter(lambda x:x>5) |
| flatMap | 将每个元素生成一个Array后压平 多用于词频统计结合split |
| sample | 对原rdd在每个分区按照比例进行抽样,第一个参数设置是否可以重复抽样 |
| distinct | 去重 |
| subtract | 找到属于前一个rdd而不属于后一个rdd的元素 a.subtract(b) |
| union | 并集 |
| intersection | 交集 |
| cartesian | 笛卡尔积 |
| sortBy | 排序 |
| zip | 按照拉链的方式连接两个RDD，类似python的zip函数 要求两个RDD具有相同的分区，每个分区元素数量相同 |
| zipWithIndex | 从0开始的递增序列按照拉链方式连接 |
PairRDD是指数据长度为2的tuple，类型
| reduceByKey | 对相同的key对应的value应用于二元聚合操作 rdd.reduceByKey(lambda x,y:x+y).collect() |
|---|---|
| groupByKey | 分组操作，得到iterator |
| sortByKey | 按照key排序,可以指定是否降序 rdd.sortByKey(False).collect() |
| join | 相当于根据key进行内连接 |
| leftOuterJoin和rightOuterJoin | 类似于表的左外和右外连接 |
| cogroup | 对两个输入分别goupByKey然后再对结果进行groupByKey |
| subtractByKey | 去除x中那些key也在y中的元素 x.subtractByKey(y).collect() |
| foldByKey | 操作和reduceByKey类似，但是要提供一个初始值 x.foldByKey(1,lambda x,y:x*y).collect() |
当一个rdd被多个任务用作中间变量时，对其进行cache缓存到内存中，加快计算。
声明对一个rdd进行cache后，该rdd不会被立即缓存，而是等到它第一次被计算出来时才进行缓存。
可以使用persist明确指定存储级别，常用的存储级别是MEMORY_ONLY和EMORY_AND_DISK。
unpersist释放缓存，unpersist是立即执行的
如果要切断血缘关系，可以用checkpoint设置检查点将某个rdd保存到磁盘中。
声明对一个rdd进行checkpoint后，该rdd不会被立即保存到磁盘，而是等到它第一次被计算出来时才保存成检查点
| cache | cache缓存到内存中，使用存储级别 MEMORY_ONLY MEMORY_ONLY意味着如果内存存储不下，放弃存储其余部分，需要时重新计算。 |
|---|---|
| persist | 缓存到内存或磁盘中，默认使用存储级别MEMORY_AND_DISK MEMORY_AND_DISK意味着如果内存存储不下，其余部分存储到磁盘中 persist可以指定其它存储级别，cache相当于persist(MEMORY_ONLY) a.unpersist() #立即释放缓存 |
| checkpoint | 将数据设置成检查点，写入到磁盘中 |
| unpersist | 释放内存 |
#checkpoint 将数据设置成检查点，写入到磁盘中。
sc.setCheckpointDir("./data/checkpoint/")
rdd_students = sc.parallelize(["LiLei","Hanmeimei","LiLy","Ann"],2)
rdd_students_idx = rdd_students.zipWithIndex()
#设置检查点后，可以避免重复计算，不会因为zipWithIndex重复计算触发不一致的问题
rdd_students_idx.checkpoint()
rdd_students_idx.take(3)
5. 共享变量
当spark集群在许多节点上运行一个函数时，默认情况下会把这个函数涉及到的对象在每个节点生成一个副本。但是，有时候需要在不同节点或者节点和Driver之间共享变量。
Spark提供两种类型的共享变量，广播变量和累加器。
broadcast
广播变量是不可变变量，实现在不同节点不同任务之间共享数据。广播变量在每个机器上缓存一个只读的变量，而不是为每个task生成一个副本，可以减少数据的传输。
accumulator
累加器主要是不同节点和Driver之间共享变量，只能实现计数或者累加功能。
累加器的值只有在Driver上是可读的，在节点上不可见
广播变量
累加器
6.分区操作
分区操作一般是改变分区操作和转换操作
| 算子 | 说明 |
|---|---|
| glom | 将一个分区内的数据转换为一个列表作为一行 |
| coalesce | shuffle可选，默认为False情况下窄依赖，不能增加分区。repartition和partitionBy调用它实现 |
| repartition | 按随机数进行shuffle，相同key不一定在同一个分区 |
| partitionBy | 按key进行shuffle，相同key放入同一个分区 |
| HashPartitioner | 默认分区器，根据key的hash值进行分区， 相同的key进入同一分区，效率较高，key不可为Array. |
| RangePartitioner | 只在排序相关函数中使用，除相同的key进入同一分区，相邻的key也会进入同一分区，key必须可排序。 |
| TaskContext | 获取当前分区id方法 TaskContext.get.partitionId |
| mapPartitions | 次处理分区内的一批数据，适合需要分批处理数据的情况，比如将数据插入某个表，每批数据只需要开启一次数据库连接，大大减少了连接开支 |
| mapPartitionsWithIndex | 类似mapPartitions，提供了分区索引，输入参数为（i，Iterator） |
| foreachPartition | 类似foreach，但每次提供一个Partition的一批数据 |
