spark之RDD
目录
2.4、Transformation与Action的区别及相关函数
2.4.1、Transformation与Action的相关函数
1、生成RDD
1.1、RDD的3种创建方式
- 第一种：读取一个外部数据集。比如，从本地文件加载数据集，或者从HDFS文件系统、HBase、Cassandra、Amazon S3等外部数据源中加载数据集。Spark可以支持文本文件、SequenceFile文件（Hadoop提供的 SequenceFile是一个由二进制序列化过的key/value的字节流组成的文本存储文件）和其他符合Hadoop InputFormat格式的文件。
- 第二种：调用SparkContext的parallelize方法，在Driver中一个已经存在的集合（数组）上创建。
- 第三种：从父RDD生成子RDD
1.2、从文件创建RDD
- （1）如果使用了本地文件系统的路径，那么，必须要保证在所有的worker节点上，也都能够采用相同的路径访问到该文件，比如，可以把该文件拷贝到每个worker节点上，或者也可以使用网络挂载共享文件系统。
- （2）textFile()方法的输入参数，可以是文件名，也可以是目录，也可以是压缩文件等。比如，textFile(“/my/directory”), textFile(“/my/directory/.txt”), and textFile(“/my/directory/.gz”).
- （3）textFile()方法也可以接受第2个输入参数（可选），用来指定分区的数目。默认情况下，Spark会为HDFS的每个block创建一个分区（HDFS中每个block默认是128MB）。你也可以提供一个比block数量更大的值作为分区数目，但是，你不能提供一个小于block数量的值作为分区数目。
eg:
1.3、通过并行集合(列表)创建RDD
eg:
nums = [1,2,3,4,5]
rdd = sc.parallelize(nums)
1.4、从父RDD生成子RDD
在RDD发生转移操作时，生成新的RDD
2、RDD的3种运算
转换（Transformation）： 基于现有的数据集创建一个新的数据集。
行动（Action）：在数据集上进行运算，返回计算值。
持久化(persist):通过持久化（缓存）机制避免这种重复计算的开销
2.1、转换（Transformation）
对于RDD而言，每一次转换操作都会产生不同的RDD，供给下一个“转换”使用。转换得到的RDD是惰性求值的，也就是说，整个转换过程只是记录了转换的轨迹，并不会发生真正的计算，只有遇到行动操作时，才会发生真正的计算，开始从血缘关系源头开始，进行物理的转换操作。
下面列出一些常见的转换操作（Transformation API）：
* filter(func)：筛选出满足函数func的元素，并返回一个新的数据集
* map(func)：将每个元素传递到函数func中，并将结果返回为一个新的数据集
* flatMap(func)：与map()相似，但每个输入元素都可以映射到0或多个输出结果
* groupByKey()：应用于(K,V)键值对的数据集时，返回一个新的(K, Iterable)形式的数据集
* reduceByKey(func)：应用于(K,V)键值对的数据集时，返回一个新的(K, V)形式的数据集，其中的每个值是将每个key传递到函数func中进行聚合
2.2、行动（Action）
行动操作是真正触发计算的地方。Spark程序执行到行动操作时，才会执行真正的计算，从文件中加载数据，完成一次又一次转换操作，最终，完成行动操作得到结果。
下面列出一些常见的行动操作（Action API）：
* count() 返回数据集中的元素个数
* collect() 以数组的形式返回数据集中的所有元素
* first() 返回数据集中的第一个元素
* take(n) 以数组的形式返回数据集中的前n个元素
* reduce(func) 通过函数func（输入两个参数并返回一个值）聚合数据集中的元素
* foreach(func) 将数据集中的每个元素传递到函数func中运行* （我的spark是2.1，报错，目前不知道啥原因）
2.3、持久化
在Spark中，RDD采用惰性求值的机制，每次遇到行动操作，都会从头开始执行计算。如果整个Spark程序中只有一次行动操作，这当然不会有什么问题。但是，在一些情形下，我们需要多次调用不同的行动操作，这就意味着，每次调用行动操作，都会触发一次从头开始的计算。这对于迭代计算而言，代价是很大的，迭代计算经常需要多次重复使用同一组数据。
2.4、Transformation与Action的区别及相关函数
2.4.1、Transformation与Action的相关函数
2.4.2、转移与执行的区别
3、spark RDD入门实例
3.1、简单的单词统计
from pyspark import SparkContext
sc = SparkContext( 'local', 'test')
#在本地进行操作
textFile = sc.textFile("file:///home/jsy/spark_test/test.txt")
wordCount = textFile.flatMap(lambda line: line.split(" ")).map(lambda word: (word,1)).reduceByKey(lambda a, b : a + b)
'''
flatMap(labmda line : line.split(” “))会遍历textFile中的每行文本内容，当遍历到其中一行文本内容时，会把文本内容赋值给变量line，
并执行Lamda表达式line : line.split(” “)。line : line.split(” “)是一个Lamda表达式，左边表示输入参数，右边表示函数里面执行的处理逻辑，
这里执行line.split(” “)，也就是针对line中的一行文本内容，采用空格作为分隔符进行单词切分，从一行文本切分得到很多个单词构成的单词集合。
'''
'''
map(lambda word : (word, 1))
针对输入的word，构建得到一个tuple，形式为(word,1)，key是word，value是1（表示该单词出现1次）
'''
'''
reduceByKey(lambda a, b : a + b)
这个操作会把所有RDD元素按照key进行分组，然后使用给定的函数
RDD的每个元素是(key,value)形式的tuple
'''
wordCount.saveAsTextFile("file:///home/jsy/spark_test/test_count")
wordCount.take(3)3.2、基本函数使用
map函数使用
#统计某行中包含的最大单词数map reduce
textFile.map(lambda line : len(line.split(" "))).reduce(lambda a,b : (a > b and a or b))
filter函数使用
#filter函数使用；统计test.txt中包含'm'字符的行
textFile.filter(lambda line : "m" in line).count()reduce函数使用
#reduce输入两个参数并返回一个值）聚合数据集中的元素
#reduce基本应用
nums = [1,2,3,4,5]
rdd = sc.parallelize(nums)
#RDD是弹性分布式数据集，通常RDD很大，会被分成很多个分区，分别保存在不同的节点上。RDD分区的一个分区原则是使得分区的个数尽量等于集群中的CPU核心（core）数目。
rdd = sc.parallelize(nums,2) #设置两个分区,后面的2是分区数
#RDD中包含了1,2,3,4,5，那么，执行reduce操作时，首先取出1和2，把a赋值为1，把b赋值为2，然后，执行大小判断，保留2。下一次，让保留下来的2赋值给a，再从RDD[Int]中取出下一个元素3，把3赋值给b，然后，对a和b执行大小判断，保留较大者3.依此类推。最终，reduce()操作会得到最大值是5。
max_=rdd.reduce(lambda a,b :(a>b and a or b))
print(max_)
total=rdd.reduce(lambda a,b:a+b)
print(total)cache实现持久化
#持久化
#rdd.cache() 实现持久化
list = ["Hadoop","Spark","Hive"]
rdd = sc.parallelize(list)
rdd.cache() #会调用persist(MEMORY_ONLY)，但是，语句执行到这里，并不会缓存rdd，这是rdd还没有被计算生成
print(rdd.count()) #第一次行动操作，触发一次真正从头到尾的计算，这时才会执行上面的rdd.cache()，把这个rdd放到缓存中
print(','.join(rdd.collect())) #第二次行动操作，不需要触发从头到尾的计算，只需要重复使用上面缓存中的rdd参考文献
上一篇： 螺旋矩阵（java实现）
下一篇： Spark之PySpark使用