from pyspark import SparkConf, SparkContext
conf = SparkConf().setAppName('RDD2').setMaster('local[2]')
sc = SparkContext(conf=conf)
print sc.version
2.0.2
sortBy
sortBy(keyfunc, ascending=True, numPartitions=None)
Sorts this RDD by the given keyfunc
x = sc.parallelize(['wills', 'kris', 'april', 'chang'])
def sortByFirstLetter(s): return s[0]
def sortBySecondLetter(s): return s[1]
y = x.sortBy(sortByFirstLetter).collect()
yy = x.sortBy(sortBySecondLetter).collect()
print '按第一个字母排序结果： {}'.format(y)
print '按第二个字母排序结果：{}'.format(yy)
按第一个字母排序结果： ['april', 'chang', 'kris', 'wills']
按第二个字母排序结果：['chang', 'wills', 'april', 'kris']
cartesian
cartesian(other)
Return the Cartesian product of this RDD and another one, that is, the RDD of all pairs of elements (a, b) where a is in self and b is in other.
rdd1 = sc.parallelize([1,2,3])
rdd2 = sc.parallelize([11,22,33])
res = rdd1.cartesian(rdd2)
print '笛卡尔结果：{}'.format(res)
笛卡尔结果：org.apache.spark.api.java.JavaPairRDD@5af637a2
print '笛卡尔结果：{}'.format(res.collect())
笛卡尔结果：[(1, 11), (1, 22), (1, 33), (2, 11), (3, 11), (2, 22), (2, 33), (3, 22), (3, 33)]
groupBy
groupBy(f, numPartitions=None, partitionFunc=func)
Return an RDD of grouped items.
rdd = sc.parallelize([1, 1, 2, 3, 5, 8])
result = rdd.groupBy(lambda x: x % 2).collect()
res = sorted([(x, sorted(y)) for (x, y) in result])
print 'groupBy后的结果：{}'.format(result)
print '优化后的结果对比：{}'.format(res)
groupBy后的结果：[(0, <pyspark.resultiterable.ResultIterable object at 0x7f79600a9990>), (1, <pyspark.resultiterable.ResultIterable object at 0x7f79600a9dd0>)]
优化后的结果对比：[(0, [2, 8]), (1, [1, 1, 3, 5])]
pipe
pipe(command, env=None, checkCode=False)
Return an RDD created by piping elements to a forked external process.
Parameters: checkCode – whether or not to check the return value of the shell command.
rdd = sc.parallelize(['wills', 'kris', 'april'])
rdd2 = rdd.pipe('grep -i "r"')
print '经过pipe处理过后的数据：{}'.format(rdd2.collect())
print rdd.pipe('grep "W"').collect()
经过pipe处理过后的数据：[u'kris', u'april']
[]
foreach
foreach(f)
Applies a function to all elements of this RDD.
def f(x): print(x)
sc.parallelize([1, 2, 3, 4, 5]).foreach(f)
max, min, sum, count
x = sc.parallelize([1, 2, 3, 4, 5])
print '最大值：{}'.format(x.max())
print '最小值：{}'.format(x.min())
print '总和：{}'.format(x.sum())
print '总个数：{}'.format(x.count())
最大值：5
最小值：1
总和：15
总个数：5
mean, variance, sampleVariance, stdev, sampleStdev
x = sc.parallelize([1, 2, 3, 4, 5])
print '平均值：{}'.format(x.mean())
print '方差：{}'.format(x.variance())
print '样本方差：{}'.format(x.sampleVariance())
print '总体标准偏差：{}'.format(x.stdev())
print '样本标准偏差：{}'.format(x.sampleStdev())
平均值：3.0
方差：2.0
样本方差：2.5
总体标准偏差：1.41421356237
样本标准偏差：1.58113883008
countByKey, countByValue
countByKey()
Count the number of elements for each key, and return the result to the master as a dictionary.
countByValue()
Return the count of each unique value in this RDD as a dictionary of (value, count) pairs.
rdd = sc.parallelize([("a", 1), ("b", 1), ("a", 1),("b", 2), ("a", 2)])
print '按key计算：{0}'.format(sorted(rdd.countByKey().items()))
print '按value计算：{0}'.format(sorted(sc.parallelize([1, 2, 1, 2, 2], 2).\
countByValue().items()))
按key计算：[('a', 3), ('b', 2)]
按value计算：[(1, 2), (2, 3)]
first, top, take, takeOrdered
first()
Return the first element in this RDD.
top(num, key=None)
Get the top N elements from a RDD.
take(num)
Take the first num elements of the RDD.
takeOrdered(num, key=None)
Get the N elements from a RDD ordered in ascending order or as specified by the optional key function.
x = sc.parallelize(range(20))
print '第一个数：{}'.format(x.first())
print '前几个数(默认降序)：{}'.format(x.top(3))
print '取几个数：{}'.format(x.take(5))
print '按一定的排序规则取数：{}'.format(x.takeOrdered(3, key=lambda x: -x))
第一个数：0
前几个数(默认降序)：[19, 18, 17]
取几个数：[0, 1, 2, 3, 4]
按一定的排序规则取数：[19, 18, 17]
collectAsMap, keys, values
collectAsMap()
Return the key-value pairs in this RDD to the master as a dictionary.
rdd = sc.parallelize([('wills', 2),('chang',4), ('kris',28)])
res = rdd.collectAsMap()
print 'map结果为：{}'.format(res)
print 'keys:{}'.format(rdd.keys().collect())
print 'values:{}'.format(rdd.values().collect())
map结果为：{'wills': 2, 'chang': 4, 'kris': 28}
keys:['wills', 'chang', 'kris']
values:[2, 4, 28]
