弹性分布式数据集(RDD)
Spark是以RDD概念为中心运行的。RDD是一个容错的、可以被并行操作的元素集合。创建一个RDD有两个方法:在你的驱动程序中并行化一个已经存在的集合;从外部存储系统中引用一个数据集。RDD的一大特性是分布式存储,分布式存储在最大的好处是可以让数据在不同工作节点并行存储,以便在需要数据时并行运算。弹性指其在节点存储时,既可以使用内存,也可已使用外存,为使用者进行大数据处理提供方便。除此之外,RDD的另一大特性是延迟计算,即一个完整的RDD运行任务被分为两部分:Transformation和Action
1.Transformation
Transformation用于对RDD的创建,RDD只能使用Transformation创建,同时还提供大量操作方法,包括map,filter,groupBy,join等,RDD利用这些操作生成新的RDD,但是需要注意,无论多少次Transformation,在RDD中真正数据计算Action之前都不可能真正运行。
2.Action
Action是数据执行部分,其通过执行count,reduce,collect等方法真正执行数据的计算部分。实际上,RDD中所有的操作都是Lazy模式进行,运行在编译中不会立即计算最终结果,而是记住所有操作步骤和方法,只有显示的遇到启动命令才执行。这样做的好处在于大部分前期工作在Transformation时已经完成,当Action工作时,只需要利用全部自由完成业务的核心工作。
下面是在python中对RDD的生成,以及一些基本的Transformation,Action操作。
# -*- coding:utf-8 -*- from pyspark import SparkContext, SparkConf from pyspark.streaming import StreamingContext import math appName ="jhl_spark_1" #你的应用程序名称 master= "local"#设置单机 conf = SparkConf().setAppName(appName).setMaster(master)#配置SparkContext sc = SparkContext(conf=conf) # parallelize:并行化数据,转化为RDD data = [1, 2, 3, 4, 5] distData = sc.parallelize(data, numSlices=10) # numSlices为分块数目,根据集群数进行分块 # textFile读取外部数据 rdd = sc.textFile("./c2.txt") # 以行为单位读取外部文件,并转化为RDD print rdd.collect() # map:迭代,对数据集中数据进行单独操作 def my_add(l): return (l,l) data = [1, 2, 3, 4, 5] distData = sc.parallelize(data) # 并行化数据集 result = distData.map(my_add) print (result.collect()) # 返回一个分布数据集 # filter:过滤数据 def my_add(l): result = False if l > 2: result = True return result data = [1, 2, 3, 4, 5] distData = sc.parallelize(data)#并行化数据集,分片 result = distData.filter(my_add) print (result.collect())#返回一个分布数据集 # zip:将两个RDD对应元素组合为元组 x = sc.parallelize(range(0,5)) y = sc.parallelize(range(1000, 1005)) print x.zip(y).collect() #union 组合两个RDD print x.union(x).collect() # Aciton操作 # collect:返回RDD中的数据 rdd = sc.parallelize(range(1, 10)) print rdd print rdd.collect() # collectAsMap:以rdd元素为元组,以元组中一个元素作为索引返回RDD中的数据 m = sc.parallelize([(‘a‘, 2), (3, 4)]).collectAsMap() print m[‘a‘] print m[3] # groupby函数:根据提供的方法为RDD分组: rdd = sc.parallelize([1, 1, 2, 3, 5, 8]) def fun(i): return i % 2 result = rdd.groupBy(fun).collect() print [(x, sorted(y)) for (x, y) in result] # reduce:对数据集进行运算 rdd = sc.parallelize(range(1, 10)) result = rdd.reduce(lambda a, b: a + b) print result
除上述以外,对RDD还存在一些常见数据操作如:
name()返回rdd的名称
min()返回rdd中的最小值
sum()叠加rdd中所有元素
take(n)取rdd中前n个元素
count()返回rdd的元素个数
更多操作请参考 :http://spark.apache.org/docs/latest/api/python/index.html