电子商务网站网络拓扑,邯郸网站建设安联网络公司,广西建设工程协会网站查询,网上商城 网站1、spark概述
spark是专为大规模数据处理而设计的快速通用计算引擎#xff0c;与Hadoop的MapReduce功能类似#xff0c;但它是基于内存的分布式计算框架#xff0c;存储还是采用HDFS。
MapReduce和Spark的区别
MapReduce的MapReduce之间需要通过磁盘进行数据传递#xf…1、spark概述
spark是专为大规模数据处理而设计的快速通用计算引擎与Hadoop的MapReduce功能类似但它是基于内存的分布式计算框架存储还是采用HDFS。
MapReduce和Spark的区别
MapReduce的MapReduce之间需要通过磁盘进行数据传递Spark直接存在内存中所以速度更快。MapReduce的Task调度和启动开销大而Spark的Task在线程中开销小一些。MapReduce编程不够灵活Spark的API丰富。MapReduce的Map和Reduce都要一次shuffle而Spark可以减少shuffle。
两者框架的区别
功能Hadoop组件Spark组件批处理MapReduce、Hive或者PigSpark Core、Spark SQL交互式计算Impala、prestoSpark SQL流式计算StormSpark Streaming机器学习MahoutSpark ML、Spark MLLib
Spark具有以下优点
基于内存速度快Java、Python和R语言可以开发spark易用性好spark框架组件丰富通用性高可以运行在多种存储结构上兼容性高。
Spark的缺点
内存消耗大。
2、Spark数据集
Spark的数据集合采用RDD(Resilient Distributed Dataset)弹性分布式数据集它是一个不可变、可分区和可并行计算的集合。
不可变RDD1到RDD2时RDD1任然存在可分区可分为多个partition并行计算Dataset是指数据集主要用于存放数据Distributed是指分布式存储并且可以进行分布式计算Resilient弹性的特点 数据可以保存在磁盘中也可以在内存中数据分布式存储也是弹性的 RDD分在多个节点上存储与HDFS的分布式存储原理类似HDFS文件以128M为基准切分为多个block存储在各个节点上而RDD则会被切分为多个partition这些partition在不同的节点上spark读取HDFS时会把HDFS上的block读到内存上对应为partitionspark计算结束时会把数据存储到HDFS上可以对应到Hive或者HBase上以HDFS为例RDD的每一个partition的大小小于128M时一个partition对应HDFS的block大于128M时则会切分为多个block。
3、RDD的数据操作
RDD的数据操作也叫做算子一共包括三类算子transformation、action和persist其中前两种进行数据处理persist进行数据存储操作。
transformation是将一个已经存在的数据集转化为一个新的数据集map就是一个transformation操作把数据集的每一个元素传给函数并返回新的RDDaction获取数据进行运算后的结果reduce就是一个action操作一般聚合RDD所有元素的操作并返回最终计算结果。persist缓存数据可以把数据缓存在内存上也可以缓存在磁盘上甚至可以到磁盘其他节点上。
我们要了解所有的transformation的操作都是lazy即不会立刻计算结果而是记录下数据集的transformation操作只有调用了action操作之后才会计算所有的transformation这样会让spark运行效率更高。 pyspark启动 进入SPARK_HOME/sbin⽬录下执⾏
pysparksparkUI 可以在spark UI中看到当前的Spark作业 在浏览器访问当前centos的4040端⼝192.168.19.137:4040 启动RDD
3.1 transformation算子
map(func):将func函数作用到数据集的每一个元素上返回一个新的RDD
rdd1 sc.parallelize([1,2,3,4,5,6,7,8,9],3)
rdd2 rdd1.map(lambda x:x1)
print(rdd2.collect())[2, 3, 4, 5, 6, 7, 8, 9, 10] filter(func):筛选func函数中为true的元素返回一个新的RDD
rdd1 sc.parallelize([1,2,3,4,5,6,7,8,9],3)
rdd2 rdd1.map(lambda x:x*2)
rdd3 rdd2.filter(lambda x:x10)
print(rdd3.collect())[12, 14, 16, 18] flatMap(func):先执行map操作然后将所有对象合并为一个对象
rdd1 sc.parallelize([a b c,d e f,h i j])
rdd2 rdd1.flatMap(lambda x:x.split( ))
rdd3 rdd1.map(lambda x:x.split( ))
print(flatmap,rdd2.collect())
print(map,rdd3.collect())flatmap [‘a’, ‘b’, ‘c’, ‘d’, ‘e’, ‘f’, ‘h’, ‘i’, ‘j’] map [[‘a’, ‘b’, ‘c’], [‘d’, ‘e’, ‘f’], [‘h’, ‘i’, ‘j’]] union(rdd):两个RDD并集
rdd1 sc.parallelize([(a, 1), (b, 2)])
rdd2 sc.parallelize([(c, 1), (b, 3)])
rdd3 rdd1.union(rdd2)
print(rdd3.collect())[(‘a’, 1), (‘b’, 2), (‘c’, 1), (‘b’, 3)] intersection(rdd):两个RDD求交集
rdd1 sc.parallelize([(a, 1), (b, 2)])
rdd2 sc.parallelize([(c, 1), (b, 3)])
rdd3 rdd1.union(rdd2)
rdd4 rdd3.intersection(rdd2)
print(rdd4.collect())[(‘c’, 1), (‘b’, 3)] groupByKey():以元祖中的第0个元素为key进行分组返回新的RDD返回的结果中value是Iterable需要list进行转化
rdd1 sc.parallelize([(a, 1), (b, 2)])
rdd2 sc.parallelize([(c, 1), (b, 3)])
rdd3 rdd1.union(rdd2)
rdd4 rdd3.groupByKey()
print(rdd4.collect())
print(list(rdd4.collect()[0][1]))[(‘b’, pyspark.resultiterable.ResultIterable object at 0x7f23ab41a4a8), (‘c’, pyspark.resultiterable.ResultIterable object at 0x7f23ab41a4e0), (‘a’, pyspark.resultiterable.ResultIterable object at 0x7f23ab41a438)] [2, 3] reduceByKey(func):将key相同的键值对按照func进行计算返回新的RDD
rdd sc.parallelize([(a, 1), (b, 1), (a, 1)])
rdd2 rdd.reduceByKey(lambda x,y:xy)
print(rdd2.collect())[(‘a’, 2), (‘b’, 1)] sortByKey(ascendingTrue, numPartitionsNone, keyfunc
tmp2 [(Mary, 1), (had, 2), (a, 3), (little, 4), (lamb,5)]
tmp2.extend([(whose, 6), (fleece, 7), (was, 8), (white,9)])
rdd1 sc.parallelize(tmp2)
rdd2 rdd1.sortByKey(True,3,keyfunclambda k:k.lower())
print(rdd2.collect())[(‘a’, 3), (‘fleece’, 7), (‘had’, 2), (‘lamb’, 5), (‘little’, 4), (‘Mary’, 1), (‘was’, 8), (‘white’, 9), (‘whose’, 6)] mapPatitions(func):分块进行map默认的map是一行行数据进行该函数是一块块进行的适合数据量大的情况。sparkContext.broadcast(要共享的数据):当某个数据需要反复查询时不用把数据放进task中可以通过⼴播变量, 通知当前worker上所有的task, 来共享这个数据,避免数据的多次复制,可以⼤⼤降低内存的开销。
3.2 action算子
collect():返回⼀个listlist中包含 RDD中的所有元素建议数量较小时使用数据较大不会全部显示
rdd1 sc.parallelize([1,2,3,4,5])
print(rdd1.collect())[1, 2, 3, 4, 5] reduce(func):将RDD中元素两两传递给输⼊函数同时产⽣⼀个新的值新产⽣的值与RDD中下⼀个元素再被传递给输⼊函数直到最后只有⼀个值为⽌。
rdd1 sc.parallelize([1,2,3,4,5])
result rdd1.reduce(lambda x,y:xy)
print(result)15 first():返回RDD中的第一个元素
rdd1 sc.parallelize([1,2,3,4,5])
result rdd1.first()
print(result)1 take(num):返回RDD的前num个元素
rdd1 sc.parallelize([1,2,3,4,5])
result rdd1.take(3)
print(result)[1, 2, 3] count():返回RDD元素个数
rdd1 sc.parallelize([1,2,3,4,5])
result rdd1.count()
print(result)5 4、Spark架构 Client客户端进程Driver一个Spark作业负责一个Driver进程负责向Master注册和注销包括StageScheduler、TaskSchedule和DAGSchedule。 StageSchedule负责生成Stage。 Stage一个Spark作业一般包含一到多个Stage。 DAGSchedule负责将Spark作业分解成一个多个Stage将Stage根据RDD的Partition个数决定Task个数然后放到TaskSchedule中。TaskSchedule将Task分配到ExecutorBackend上执行并监控Task状态。 Task一个Stage包含一个多个Task多个Task实现并行运行。 ApplicationSpark应用程序批处理作业的集合。其中main方法时入口定义了RDD和RDD的操作。MasterStandalone模式中的主控节点负责接收Client提交的作业管理Worker并让Worker启动Driver和Executro。WorkerStandalone模式中的salve节点上的守护节点负责管理本节点的资源定期向Master汇报心跳接收Master命令启动Driver和Executor。
Spark作业的Stage划分
窄依赖父RDD的每个Partition最多被一个子RDD的Partition所使用即一个父RDD对应一个子RDD。map、filter、union、join对输入做协同划分。宽依赖子RDD依赖所有父RDD分区。groupByKey、join对输入做非协同划分。
窄依赖的所有RDD作为一个Stage遇到宽依赖结束。
