spark核心知识点总结

柔光的暖阳◎ 2022-02-18 18:47 278阅读 0赞

# 特点 #

*  大规模快速通用的计算引擎（hadoop 花费90%时间用户rw）
 *  DAG：有向无环图
 *  使用80+算子（操作符），容易构建并行应用
 *  基于Hadoop的mr，扩展mr模型高效实用，内存型集群计算，提高app处理速度。
 *  支持语言多

# spark模块 #

core				//通用执行引擎，提供内存计算和对外部数据集的引用。
    SQL					//构建在core之上，引入新的抽象SchemaRDD，提供了结构化和半结构化支持。
    Streaming			//小批量计算，RDD.
    MLlib				//机器学习库。core在。
    Graphx				//图计算。

# 初识api #

**sc对象**

*  sparkContext spark程序入口点，封装了整个spark运行环境的信息，代表了到spark集群的连接

**sparkConf**

*  配置对象

**RDD**

*  resilient distributed dataset，弹性分布式数据集 //等价于 集合。

# RDD #

*  是spark的基本数据结构，是不可变（只读）数据集。
 *  RDD中的数据集进行逻辑分区，每个分区可以单独在集群节点并行计算。
 *  RDD是只读的记录分区集合。RDD具有容错机制。
 *  内存处理计算（在job间进行数据共享。内存的IO速率高于网络和磁盘。）  
    　　*分布式内存：各个主机之间开辟的内存空间，地址是相同区域*

**RDD的容错** ：

*  RDD是一个不可变的、确定性的可重复计算的分布式数据集
 *  RDD的某些partition丢失了，可以通过血统（lineage）信息重新计算恢复；
 *  如果RDD任何分区因worker节点故障而丢失，那么这个分区可以从原来依赖的容错数据集中恢复；
 *  由于Spark中所有的数据的转换操作都是基于RDD的，即使集群出现故障，只要输入数据集存在，所有的中间结果都是可以被计算的。

**内部包含5个主要属性**  
　1.分区列表（分区决定了并行度）  
　2.针对每个切片的计算函数。  
　3.对其他rdd的依赖列表  
　4.可选，如果是KeyValueRDD的话，可以带分区类。  
　5.可选，首选块位置列表(hdfs block location);

**RDD持久化**  
通过调用cache（）和persist()方法实现

*  区别：后者是前者的底层无参或persist（MEMOEY\_ONLY）版本实现。
 *  清除缓存调用unpersist（）方法
 *  持久化级别：内存，串行化内存，磁盘，磁盘+内存等

RDD变换

*  返回指向新rdd的指针，在rdd之间创建依赖关系。
 *  每个rdd都有计算函数和指向父RDD的指针。
 *  spark是延迟计算，只有执行动作的时候才运行。
 *  分辨变换和动作：方法返回值是rdd就是变换

# spark源码分析 #

*  rdd变换：  
    textfile（）加载文件：返回hadoopRDD（创建HadoopRDD之前，先将hadoopConfiguration进行广播）调用map方法，最终返回MapPartitionsRDD  
    ![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MzA5MzUwMQ_size_16_color_FFFFFF_t_70]  
    详细流程：[https://blog.csdn.net/weixin\_43093501/article/details/89492421][https_blog.csdn.net_weixin_43093501_article_details_89492421]  
    ![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MzA5MzUwMQ_size_16_color_FFFFFF_t_70 1]
 *  rdd提交：
    
     *  sparkcontext运行作业
     *  调用有向无环图调度器的运行作业方法
     *  有向无环图提交作业到事件轮询对象中，将其封装成事件然后发送到轮询队列。
     *  有一个线程监控队列，将事件最终发送到有向无环图进行处理。
     *  具体内容为将stage递归提交，然后将task串行化广播到集群。
     *  任务调度器将任务集提交到后台调度器。
     *  后台调度器使用netty发送套接字到消息轮询对象中。
     *  有一个线程监控，将数据发送到执行器中，执行器执行任务。
     *  执行器中有线程池，创建线程运行task。
     *  首先反序列化，然后运行，运行task中的内容（resultmapTask/shuffleMapTask）,然后执行write（）方法  
        ![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MzA5MzUwMQ_size_16_color_FFFFFF_t_70 2]  
        ![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MzA5MzUwMQ_size_16_color_FFFFFF_t_70 3]  
        【DAGScheduler是高级调度器层】：按照stage对每个Job的各阶段计算有向无环图，跟踪rdd和每个stage的输出。找出最小调度，运行作业，将Stage对象以TaskSet方式交给底层调度器。底层调度器实现TaskScheduler，进而在集群上运行job。  
        ![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MzA5MzUwMQ_size_16_color_FFFFFF_t_70 4]

【Task】：一般来说，一个 rdd 有多少个 partition，就会有多少个 task，因为每一个 task 只是处理一个 partition 上的数据。

【Stage】：task的集合，在shuffle的边界进行隔离（划分）。

*  ShuffleMapStage：产生输出数据，在每次shuffle之前发生。可共享
 *  ResultStage：用于执行action动作的最终stage

【job】:提交给调度的顶层的工作项,是Stage集合。

*  result job，计算ResultStage来执行action.
 *  map-state job,为shuffleMapState结算计算输出结果以供下游stage使用。

# 依赖（dependency） #

窄依赖：  
NarrowDependency:

*  子RDD的每个分区依赖于父RDD的少量分区

宽依赖：  
ShuffleDependency

*  子RDD的每个分区依赖于父RDD的多个分区

# SparkSQL #

该模块能在spark运行sql语句。  
DataFrame 数据框（相当于数据库中的表）  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MzA5MzUwMQ_size_16_color_FFFFFF_t_70 5]

### sparkSQL使用jdbc(java代码) ###

public class SqlJDBC { 
        public static void main(String[] args) { 
            SparkConf conf = new SparkConf();
            conf.setMaster("local");
            conf.setAppName("SqlJDBC");
            SparkSession session = SparkSession.builder().
                                    appName("SqlJDBC").
                                    config("spark.master", "local").
                                    getOrCreate();
            String url = "jdbc:mysql://localhost:3306/dbtest";
            String table = "student";
            Dataset<Row> df = session.read().format("jdbc")
                                .option("url",url)
                                .option("dbtable",table)
                                .option("user","root")
                                .option("password","root")
                                .option("driver","com.mysql.jdbc.Driver")
                                .load();
            df.show();
    // //投影查询
            Dataset<Row> df2 = df.select(new Column("id"));
            df2.show();
    // //过滤
            df2 = df2.where("age like '%3'");
    // //去重
            df2 = df2.distinct();
    
            //写入
            Properties prop = new Properties();
            prop.put("user", "root");
            prop.put("password", "root");
            prop.put("driver", "com.mysql.jdbc.Driver"
            df.write().jdbc(url,"subpersons",prop);
    // df.show();
        }
    }

# spark Streaming #

*  针对实时数据流处理,具有可扩展、***高吞吐量***、容错.
 *  内部，spark接受实时数据流，分成batch(分批次)进行处，最终在每个batch终产生结果stream.（并非实时计算，而是分批次计算）

SparkStreaming接收实时数据，根据时间划分批次，形成一个个的RDD，他们会形成一个序列，称为离散流（DStream）

*  窗口长度：一个窗口覆盖的流数据的时间长度。必须是批处理时间间隔的倍数，
 *  滑动时间间隔：前一个窗口到后一个窗口所经过的时间长度。必须是批处理时间间隔的倍数

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MzA5MzUwMQ_size_16_color_FFFFFF_t_70]: /images/20220218/6514bfb783c343fca811f2466c9e1685.png
[https_blog.csdn.net_weixin_43093501_article_details_89492421]: https://blog.csdn.net/weixin_43093501/article/details/89492421
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MzA5MzUwMQ_size_16_color_FFFFFF_t_70 1]: /images/20220218/a007e57a755c4220aee62112a5001cd9.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MzA5MzUwMQ_size_16_color_FFFFFF_t_70 2]: /images/20220218/af08179ae945472799b89a28d6d0999c.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MzA5MzUwMQ_size_16_color_FFFFFF_t_70 3]: /images/20220218/9588847598584fa9a1c9b5648370772b.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MzA5MzUwMQ_size_16_color_FFFFFF_t_70 4]: /images/20220218/3f4d36c4c3e34040809e8d614addbd25.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MzA5MzUwMQ_size_16_color_FFFFFF_t_70 5]: /images/20220218/add0863af270410a9130cb63ae8e1624.png