Spark运行原理和RDD解析(DT大数据梦工厂)

Posted 2020-06-13

Spark一般基于内存，一些情况下也会基于磁盘

Spark优先会把数据放到内存中，如果内存实在放不下，也会放到磁盘里面的

不单能计算内存放的下的数据，也能计算内存放不下的数据

实际如果数据大于内存，则要考虑数据放置策略和优化算法，因为Spark初衷是一寨式处理

小到5~10台的分布式大到8000台的规模，Spark都能运行

大数据计算问题：交互式查询（基于shell、sparkSQL）、批处理、机器学习和计算等等

底层基于RDD，分布式弹性数据级，支持各种各样的比如流处理、SQL、SparkR等等范式

==========Spark特点============

要理解Spark，从以下方面理解

1、分布式多台机器运行

     不同节点会处理其中一部分数据，各个节点数据处理互不干扰，分布式做并行化

     Cluster Manager负责分配资源到各个节点中，各节点算完之后，再汇总到Cluster Manager再统一输出

     

2、内存式+磁盘计算

     例如：300万条数据，分配到3台机器，比如每台机器100万(也可能不是平均)，一台机器100万条内存放的下放内存，放不下就内存

3、迭代式计算是Spark真正的精髓

     将计算分成N个过程，一个过程结束再下一个 

     shuffle是一个节点到另外一个节点

==========开发============

我们写好程序通过Driver到各个机器

为什么大部分大部分用JAVA写程序

因为JAVA的人比较多，Scala的人比较少

JAVA和J2EE融合比较方便

后续维护比较方便

缺点：JAVA开发Spark太繁琐了

后面的例子同时使用Scala和JAVA实现

开发是单独的机器，提交机器是在另外的机器上

可以处理的数据来源：SparkWorker可以从各种数据，除了HDFS、HBase，还可以从Hive，oracle，mysql

                                 注：Hive数据仓库，数据引擎，SparkSQL可以实现这个，但是不能完全取代Hive

处理数据输出：HDFS、HBase、Hive、oracle、s3或者直接返回到客户端等等

==========运行============

一切基于RDD（弹性分布式数据集），

弹性之一：

进行数据分片，默认放在内存中的，如果内存放不下，一部分会放在磁盘上进行保存，对用户来说完全不需要关心它数据是在哪里的，RDD会自动进行内存和磁盘的切换

弹性之二：

基于Lineage的高效容错，假设一个作业步骤有1000个步骤，假设在901步骤的时候出错，它会自动从900个步骤进行纠错回复重新计算

弹性之三：

Task如果失败，会自动进行特定次数的重试，假设一个Task有1000个步骤，假设901步骤出错，如果从900个步骤开始重试，会有一定次数的重试，还是失败就真失败

弹性之四：

Stage如果失败，会自动特定次数的重试，而且只会计算失败的分片

注：Stage，其实就是阶段

==========做缓存时机 ============

1、特别耗时

2、计算链条很长

3、Shuffle之后，如果这之后失败做过缓存，就不用再做shuffle了

4、checkpoint之前，前面的步骤都缓存过了之后，如果checkpoint断了，之前的就保存了

==========RDD例子 ============

除了hadoop、spark的start-all

hadoop应该启动./start-df.sh

spark应该启动./start-

http://master:18080  看曾经运行的作业的信息

spark的bin下面./spark-shell --master spark://Master:7077

val data = sc.textFile("/library/wordcount/input/Data")或者sc.textFile("hdfs//Master:9000/library/wordcount/input/Data")

Spark自己会创建RDD

data.roDebugString看数据依赖关系

可以看出是MapPartitionnsRDD是分片方式，分布在不同机器上的

data.count看数据

http://Master:4040看job

数据不动，代码动，数据都分布在各个机器上的。

一个Block Size一般是128M，实际Partition和Block Size可能会有出入

val flatted = data.flatMap(_.spilit(" "))

又产生的了新的MapPartitionnsRDD

val mapped = flatted.map(word=>(word,1))//每个单词计数为1

val reduced = mapped.reduceByKey(_+_)

key相同的相加，产生shuffle

reduced .saveAsTextFIle("/library/wordcount/input/Data/output/onclick4")

另外：

Data2222不存在，开始加载是lazy的，所以不会报错，只有data.count就会报错

下一讲：eclipse开发JAVA和scala角度开发测试和运行程序

作业：写一篇博客，写你理解的Spark的基本理解

王家林老师名片：

中国Spark第一人

新浪微博：http://weibo.com/ilovepains

微信公众号：DT_Spark

博客：http://blog.sina.com.cn/ilovepains

手机：[phone]

QQ：1740415547

邮箱：[email protected]

本文出自 “一枝花傲寒” 博客，谢绝转载！