目录
引例入门：textFile、collect、filter、first、persist、count
创建RDD的方式：parallelize、textFile
转化操作：map、filter、flatMap、sample、union、intersection、subtract、cartesian、distinct、groupByKey、reduceByKey、sortByKey、join、cogroup
行动操作：reduce、collect、count、first、take、takeSample、takeOrdered、saveAsTextFile、saveAsSequenceFile、countByKey、foreach
简单的wordCount程序
|
引例入门textFile、collect、filter、first、persist、count |
# -*-coding=utf-8 -*- from pyspark import SparkConf, SparkContext sc = SparkContext('local') lines = sc.textFile("README.md") #使用textFile()创建一个字符串RDD。lines是一个字符串RDD（弹性分布式数据集）。文件中的每一行是其中的一个元素 print type(lines) #<class 'pyspark.rdd.RDD'> print lines.collect()[0] #collect()从RDD转换到List 打印第一个元素：# Apache Spark pythonLines = lines.filter(lambda line: "Python" in line) #filter是对RDD元素进行过滤；返回一个新的数据集，由经过func函数后返回值为true的元素组成 for i in pythonLines.collect(): print i ''' high-level APIs in Scala, Java, Python, and R, and an optimized engine that ## Interactive Python Shell Alternatively, if you prefer Python, you can use the Python shell: ''' print type(pythonLines.collect()) #<type 'list'> print pythonLines.first() #获取第一行：high-level APIs in Scala, Java, Python, and R, and an optimized engine that pythonLines.persist() #如果pythonLines被下游程序反复使用，将pythonLines持久化到内存可以提高计算效率 print pythonLines.count() #求RDD中元素个数。输出：3
|
创建RDD的方式：parallelize、textFile |
# -*-coding=utf-8 -*- from pyspark import SparkConf, SparkContext sc = SparkContext('local') print sc.parallelize([1, 2, 3, 4]).collect() #[1, 2, 3, 4] print sc.textFile("README.md").collect() #[u'# Apache Spark', u'', u'Spark is a fast and general cluster computing system for Big Data. It provides'
|
转化操作：map、filter、flatMap、sample、union、intersection、subtract、 cartesian、distinct、groupByKey、reduceByKey、sortByKey、join、cogroup、mapValues |
# -*-coding=utf-8 -*- from pyspark import SparkConf, SparkContext sc = SparkContext('local') #map(func):对RDD中的每个元素都执行一个指定的函数产生一个新的RDD。RDD之间的元素是一对一关系 print sc.parallelize([1, 2, 3, 4]).map(lambda x: x * x).collect() #[1, 4, 9, 16] print sc.parallelize(["hello world", "hi"]).map(lambda line: line.split(" ")).collect() #[['hello', 'world'], ['hi']] #filter(func):是对RDD元素进行过滤；返回一个新的数据集，由经过func函数后返回值为true的元素组成 print sc.parallelize([1, 2, 3, 4]).filter(lambda x: x>2).collect() #[3, 4] #flatMap(func):类似于map，但是输出结果会被“拍扁” print sc.parallelize(["hello world", "hi"]).flatMap(lambda line: line.split(" ")).collect() #['hello', 'world', 'hi'] #sample(withReplacement,fraction,seed)根据给定的随机种子seed，随机抽样出fraction比例的数据，withReplacement：是否放回抽样 print sc.parallelize([1, 2, 3, 4, 5, 6, 7]).sample(True,0.2,1).collect() #[1, 7, 7] #union(RDD):RDD取并集 print sc.parallelize([1, 2, 3]).union(sc.parallelize([3, 4, 5])).collect() #[1, 2, 3, 3, 4, 5] #intersection(RDD):RDD取交集 print sc.parallelize([1, 2, 3]).intersection(sc.parallelize([3, 4, 5])).collect() #[3] #subtract(RDD):差集 print sc.parallelize([1, 2, 3]).subtract(sc.parallelize([3, 4, 5])).collect() #[2, 1] #cartesian(RDD):笛卡尔乘积，作用于数据集T和U上，返回(T, U)，即数据集中每个元素的两两组合 print sc.parallelize([1, 2, 3]).cartesian(sc.parallelize([3, 4, 5])).collect() #[(1, 3), (1, 4), (1, 5), (2, 3), (2, 4), (2, 5), (3, 3), (3, 4), (3, 5)] #distinct():RDD去重 print sc.parallelize([1, 2, 3, 3]).distinct().collect() #[1, 2, 3] #groupByKey():作用于由键值对(K, V)组成的数据集上，将Key相同的数据放在一起，返回一个由键值对(K, Iterable)组成的数据集 a = sc.parallelize({(1,2),(3,4),(3,6)}).groupByKey().collect() #[(1, Iterable), (3, Iterable)] for i in a: print str(i[0])+":"+str(list(i[1])) #1:[2] ; 3:[4, 6] #reduceByKey():作用于键值对(K, V)上，按Key分组，然后将Key相同的键值对的Value都执行func操作，得到一个值 print sc.parallelize({(1,2),(3,4),(3,6)}).reduceByKey(lambda x,y: x+y).collect() #[(1, 2), (3, 10)] #sortByKey([ascending=True], [numTasks]):按照Key进行排序，ascending的值默认为True，True/False表示升序还是降序 print sc.parallelize({(2,2),(1,4),(3,6)}).sortByKey().collect() #[(1, 4), (2, 2), (3, 6)] #join(otherDataset, [numTasks]):类似于SQL中的连接操作，即作用于键值对(K, V)和(K, W)上，返回元组 (K, (V, W))，spark也支持外连接，包括leftOuterJoin，rightOuterJoin和fullOuterJoin。例子： print sc.parallelize({(1,2),(3,4),(3,6)}).join(sc.parallelize({(3,7),(4,8)})).collect() #[(3, (4, 7)), (3, (6, 7))] print sc.parallelize({(1,2),(3,4),(3,6)}).leftOuterJoin(sc.parallelize({(3,7),(4,8)})).collect() #[(1, (2, None)), (3, (4, 7)), (3, (6, 7))] print sc.parallelize({(1,2),(3,4),(3,6)}).rightOuterJoin(sc.parallelize({(3,7),(4,8)})).collect() #[(4, (None, 8)), (3, (4, 7)), (3, (6, 7))] print sc.parallelize({(1,2),(3,4),(3,6)}).fullOuterJoin(sc.parallelize({(3,7),(4,8)})).collect() #[(4, (None, 8)), (1, (2, None)), (3, (4, 7)), (3, (6, 7))] #cogroup(otherDataset, [numTasks]):作用于键值对(K, V)和(K, W)上，返回元组 (K, (Iterable, Iterable))。这一操作可叫做groupWith。 a = sc.parallelize({(1,2),(3,4),(3,6)}).cogroup(sc.parallelize({(3,7),(4,8)})).collect() #[(4, (Iterable, Iterable)), (1, (Iterable, Iterable)), (3, (Iterable, Iterable))] for i in a: print str(i[0])+":"+str(list(i[1][0]))+","+str(list(i[1][1])) #4:[],[8] ; 1:[2],[] ;3:[4, 6],[7] #mapValues(func): 扩展值 print sc.parallelize({("panda",0),("pink",3)}).mapValues(lambda x:(x,1)).collect() #[('pink', (3, 1)), ('panda', (0, 1))]
|
行动操作：reduce、collect、count、first、take、takeSample、takeOrdered、 saveAsTextFile、saveAsSequenceFile、countByKey、foreach、countByValue |
# -*-coding=utf-8 -*- from pyspark import SparkConf, SparkContext sc = SparkContext('local') #reduce(func):对数据集的所有元素执行func函数 print sc.parallelize([1, 2, 3, 3]).reduce(lambda x,y:x+y) #9 #collect():对数据集转化为List print sc.parallelize([1, 2, 3, 3]).collect() #[1, 2, 3, 3] #count(): 返回数据集元素个数 print sc.parallelize([1, 2, 3, 3]).count() #4 #first(): 返回数据集中第一个元素 print sc.parallelize([1, 2, 3, 3]).first() #4 #take(n): 返回数据集中前n个元素元素 print sc.parallelize([1, 2, 3, 3]).take(2) #4 #takeSample(withReplacementnum,[seed])返回包含随机的num个元素的数组，和sample不同，takeSample 是行动操作，所以返回的是数组而不是RDD,其中第一个参数withReplacement是抽样时是否放回，第二个参数num会精确指定抽样数，而不是比例。 print sc.parallelize([1, 2, 3, 3]).takeSample(True,2) #[1, 3] #takeOrdered(n,[ordering]): 按自然顺序或者指定的排序规则返回前n个元素 print sc.parallelize([10, 1, 2, 9, 3, 4, 5, 6, 7]).takeOrdered(6) #[1, 2, 3, 4, 5, 6] print sc.parallelize([10, 1, 2, 9, 3, 4, 5, 6, 7]).takeOrdered(6, key=lambda x: -x) #[10, 9, 7, 6, 5, 4] print sc.parallelize([(10, 1), (2, 9), (3, 4), (5, 6)]).takeOrdered(6, key=lambda x: -x[1]) #[(2, 9), (5, 6), (3, 4), (10, 1)]按第二个的降序排序 #saveAsTextFile(path): 保存为TextFile文件 sc.parallelize([1, 2, 3, 3]).saveAsTextFile("a") #[(2, 9), (5, 6), (3, 4), (10, 1)]按第二个的降序排序 #saveAsSequenceFile(path): 保存为SequenceFile文件 sc.parallelize({("11","2"),("103","4"),("103","6")}).saveAsSequenceFile("b") #[(2, 9), (5, 6), (3, 4), (10, 1)]按第二个的降序排序 #countByKey(): 对于(K,V)类型的RDD。返回一个(K,Int)的字典，Int为K的个数 print sc.parallelize([("a", 1), ("b", 1), ("a", 1)]).countByKey() #defaultdict(<type 'int'>, {'a': 2, 'b': 1}) #foreach(func): 对数据集中的每个元素执行func函数 def f(x): print(x*2) sc.parallelize([1, 2, 3, 3]).foreach(f) #2,4,6,6 #countByValue(): 根据值计数 print sc.parallelize([1, 2, 1, 2, 2]).countByValue().items() #[(1, 2), (2, 3)]
|
简单的wordCount程序 |
# -*-coding=utf-8 -*- from pyspark import SparkConf, SparkContext sc = SparkContext('local') #方法一: print sc.parallelize(["hello","hi","hello"]).flatMap(lambda x: x.split(" ")).collect() #['hello', 'hi', 'hello'] print sc.parallelize(["hello","hi","hello"]).flatMap(lambda x: x.split(" ")).map(lambda x:(x,1)).reduceByKey(lambda x,y:x+y).collect() #[('hi', 1), ('hello', 2)] #方法二: print sc.parallelize(["hello","hi","hello"]).flatMap(lambda x: x.split(" ")).collect() #['hello', 'hi', 'hello'] print sc.parallelize(["hello","hi","hello"]).flatMap(lambda x: x.split(" ")).countByValue() #defaultdict(<type 'int'>, {'hi': 1, 'hello': 2})