spark2.x由浅入深深到底系列七之RDD python api详解一

学习spark任何技术之前，请先正确理解spark，可以参考：正确理解spark

以下对RDD的三种创建方式、单类型RDD基本的transformation api、采样Api以及pipe操作进行了python api方面的阐述

一、RDD的三种创建方式

1. 从稳定的文件存储系统中创建RDD，比如local fileSystem或者hdfs等，如下：

"""
创建RDD的方法:
1: 从一个稳定的存储系统中，比如hdfs文件, 或者本地文件系统
"""
text_file_rdd = sc.textFile("file:////Users/tangweiqun/spark-course/word.txt")
print "text_file_rdd = {0}".format(",".join(text_file_rdd.collect()))

2.  可以经过transformation api从一个已经存在的RDD上创建一个新的RDD，以下是map这个转换api

"""
2: 从一个已经存在的RDD中, 即RDD的transformation api
"""
map_rdd = text_file_rdd.map(lambda line: "{0}-{1}".format(line, "test"))
print "map_rdd = {0}".format(",".join(map_rdd.collect()))

3.  从一个内存中的列表数据创建一个RDD，可以指定RDD的分区数，如果不指定的话，则取所有Executor的所有cores数量

"""
3: 从一个已经存在于内存中的列表,  可以指定分区，如果不指定的话分区数为所有executor的cores数,下面的api时指定了2个分区
"""
parallelize_rdd = sc.parallelize([1, 2, 3, 3, 4], 2)
print "parallelize_rdd = {0}".format(parallelize_rdd.glom().collect())

注：对于第三种情况，scala中还提供了makeRDD api，这个api可以指定创建RDD每一个分区所在的机器，这个api的原理详见spark core RDD scala api中

二、单类型RDD基本的transformation api

先基于内存中的数据创建一个RDD

conf = SparkConf().setAppName("appName").setMaster("local")
sc = SparkContext(conf=conf)

parallelize_rdd = sc.parallelize([1, 2, 3, 3, 4], 2)

1.  map操作，表示对parallelize_rdd的每一个元素应用我们自定义的函数接口，如下是将每一个元素加1：

map_rdd = parallelize_rdd.map(lambda x: x + 1)
"""
    结果：[[2, 3], [4, 4, 5]]
    """
print "map_rdd = {0}".format(map_rdd.glom().collect())

需要注意的是，map操作可以返回与RDD不同类型的数据，如下，返回一个String类型对象：

map_string_rdd = parallelize_rdd.map(lambda x: "{0}-{1}".format(x, "test"))
"""
    结果：[['1-test', '2-test'], ['3-test', '3-test', '4-test']]
    """
print "map_string_rdd = {0}".format(map_string_rdd.glom().collect())

2.  flatMap操作，对parallelize_rdd的每一个元素应用我们自定义的lambda函数，这个函数的输出是一个数据列表，flatMap会对这些输出的数据列表进行展平

flatmap_rdd = parallelize_rdd.flatMap(lambda x: range(x))
"""
    结果：[[0, 0, 1], [0, 1, 2, 0, 1, 2, 0, 1, 2, 3]]
    """
print "flatmap_rdd = {0}".format(flatmap_rdd.glom().collect())

3.  filter操作，对parallelize_rdd的每一个元素应用我们自定义的过滤函数，过滤掉我们不需要的元素，如下，过滤掉不等于1的元素：

filter_rdd = parallelize_rdd.filter(lambda x: x != 1)
"""
    结果：[[2], [3, 3, 4]]
    """
print "filter_rdd = {0}".format(filter_rdd.glom().collect())

4.  glom操作，查看parallelize_rdd每一个分区对应的元素数据

glomRDD = parallelize_rdd.glom()
"""
结果：[[1, 2], [3, 3, 4]] 说明parallelize_rdd有两个分区，第一个分区中有数据1和2,第二个分区中有数据3,3和4
"""
print "glomRDD = {0}".format(glomRDD.collect())

5.  mapPartitions操作，对parallelize_rdd的每一个分区的数据应用我们自定义的函数接口方法，假设我们需要为每一个元素加上一个初始值，而这个初始值的获取又是非常耗时的，这个时候用mapPartitions会有非常大的优势，如下：

//这是一个初始值获取的方法，是一个比较耗时的方法
def get_init_number(source):
    print "get init number from {0}, may be take much time........".format(source)
    time.sleep(1)
    return 1

def map_partition_func(iterator):
    """
    每一个分区获取一次初始值，integerJavaRDD有两个分区，那么会调用两次getInitNumber方法
    所以对应需要初始化的比较耗时的操作，比如初始化数据库的连接等，一般都是用mapPartitions来为对每一个分区初始化一次，而不要去使用map操作
    :param iterator:
    :return:
    """
    init_number = get_init_number("map_partition_func")
    yield map(lambda x : x + init_number, iterator)
map_partition_rdd = parallelize_rdd.mapPartitions(map_partition_func)
"""
    结果：[[[2, 3]], [[4, 4, 5]]]
    """
print "map_partition_rdd = {0}".format(map_partition_rdd.glom().collect())

def map_func(x):
    """
    遍历每一个元素的时候都会去获取初始值，这个integerJavaRDD含有5个元素，那么这个getInitNumber方法会被调用4次，严重的影响了时间，不如mapPartitions性能好
    :param x:
    :return:
    """
    init_number = get_init_number("map_func")
    return x + init_number
map_rdd_init_number = parallelize_rdd.map(map_func)
"""
    结果：[[2, 3], [4, 4, 5]]
    """
print "map_rdd_init_number = {0}".format(map_rdd_init_number.glom().collect())

6.  mapPartitionsWithIndex操作，对parallelize_rdd的每一个分区的数据应用我们自定义的函数接口方法，在应用函数接口方法的时候带上了分区信息，即知道你当前处理的是第几个分区的数据

def map_partition_with_index_func(partition_index, iterator): yield (partition_index, sum(iterator))
map_partition_with_index_rdd = parallelize_rdd.mapPartitionsWithIndex(map_partition_with_index_func)
"""
    结果：[[(0, 3)], [(1, 10)]]
    """
print "map_partition_with_index_rdd = {0}".format(map_partition_with_index_rdd.glom().collect())

三、采样Api

先基于内存中的数据创建一个RDD

conf = SparkConf().setAppName("appName").setMaster("local")
sc = SparkContext(conf=conf)

parallelize_rdd = sc.parallelize([1, 2, 3, 3, 4], 2)

1.  sample

"""
第一个参数为withReplacement
如果withReplacement=true的话表示有放回的抽样，采用泊松抽样算法实现
如果withReplacement=false的话表示无放回的抽样，采用伯努利抽样算法实现

第二个参数为：fraction，表示每一个元素被抽取为样本的概率，并不是表示需要抽取的数据量的因子
比如从100个数据中抽样，fraction=0.2，并不是表示需要抽取100 * 0.2 = 20个数据，
而是表示100个元素的被抽取为样本概率为0.2;样本的大小并不是固定的，而是服从二项分布
当withReplacement=true的时候fraction>=0
当withReplacement=false的时候 0 < fraction < 1

第三个参数为：reed表示生成随机数的种子，即根据这个reed为rdd的每一个分区生成一个随机种子
"""
sample_rdd = parallelize_rdd.sample(False, 0.5, 100)
"""
结果：[[1], [3, 4]]
"""
print "sample_rdd = {0}".format(sample_rdd.glom().collect())

2.  randomSplit

"""
//按照权重对RDD进行随机抽样切分，有几个权重就切分成几个RDD
//随机抽样采用伯努利抽样算法实现, 以下是有两个权重，则会切成两个RDD
"""
split_rdds = parallelize_rdd.randomSplit([0.2, 0.8])
print len(split_rdds)
"""[[], [3, 4]]"""
print "split_rdds[0] = {0}".format(split_rdds[0].glom().collect())
"""[[1, 2], [3]]"""
print "split_rdds[1] = {0}".format(split_rdds[1].glom().collect())

3.  takeSample

"""
//随机抽样指定数量的样本数据
//第一个参数为withReplacement
//如果withReplacement=true的话表示有放回的抽样，采用泊松抽样算法实现
//如果withReplacement=false的话表示无放回的抽样，采用伯努利抽样算法实现
//第二个参数指定多少，则返回多少个样本数
"""
"""随机抽样指定数量的样本数据
结果：[1]
"""
print parallelize_rdd.takeSample(False, 1)

4. 分层采样，对key-value类型的RDD进行采样

"""创建一个key value类型的RDD"""
pair_rdd = sc.parallelize([('A', 1), ('B', 2), ('C', 3), ('B', 4), ('A', 5)])
sampleByKey_rdd = pair_rdd.sampleByKey(withReplacement=False, fractions={'A':0.5, 'B':1, 'C':0.2})
"""
结果：[[('A', 1), ('B', 2), ('B', 4)]]
"""
print "sampleByKey_rdd = {0}".format(sampleByKey_rdd.glom().collect())

抽样的原理详细可以参考：spark core RDD api。这些原理性的东西用文字不太好表述

四、pipe，表示在RDD执行流中的某一步执行其他的脚本，比如python或者shell脚本等

conf = SparkConf().setAppName("appName").setMaster("local")
sc = SparkContext(conf=conf)

parallelize_rdd = sc.parallelize(["test1", "test2", "test3", "test4", "test5"], 2)

"""
//如果是在真实的spark集群中，那么要求echo.py在集群的每一台机器的同一个目录下面都要有
//第二个参数是环境变量
"""
pipe_rdd = parallelize_rdd.pipe("python /Users/tangweiqun/spark/source/spark-course/spark-rdd-java/src/main/resources/echo.py", {"env":"env"})
"""
结果：slave1-test1-env slave1-test2-env slave1-test3-env slave1-test4-env slave1-test5-env
"""
print "pipe_rdd = {0}".format(" ".join(pipe_rdd.collect()))

echo.py的内容如下：

import sys
import os

#input = "test"
input = sys.stdin
env_keys = os.environ.keys()
env = ""
if "env" in env_keys:
   env = os.environ["env"]
for ele in input:
   output = "slave1-" + ele.strip('/n') + "-" + env
       print (output)

input.close

对于pipe的原理，以及怎么实现的，参考：spark core RDD api，这个里面还清楚的讲述了怎么消除手工将脚本拷贝到每一台机器中的工作