Spark、RDD、算子

川长思鸟来 2021-12-03 10:21 468阅读 0赞

# Spark、RDD、算子 #

spark特点：速度快(基于内存处理数据，有相无关图)，语言广

## 一、Spark核心-RDD（弹性分布式数据） ##

Rdd是逻辑概念，partition视具体执行操作

### RDD特性 ###

1 RDD是由一系列partition组成

partition默认数量和block块数量一样

2 算子(函数)作用在RDD的partition上

3 RDD之间存在依赖关系

数据丢失可以恢复

4 分区器式作用在看k,v格式的RDD上

5 partition提供数据计算的最佳位置，利于数据化的本地处理—计算向数据移动

### RDD附 ###

1 Spark 读取HDFS中数据的方法-textFile，底层是调用的MR读取HDFS文件的方法，首先会split，每个split对应一个block，每个split对应生成RDD的每个partition

2 RDD的数据是一个个tuple2数据，这个RDD就是kv格式的RDD

3 RDD弹性

1 RDD之间的依赖关系

2 RDD 的partition数量可多可少

4 RDD的partition分布在多个节点上

## 二、算子 ##

算子分类

Transformations转换算子延迟执行 需要Action算子触发才执行

Action行动算子，触发执行

控制算子 cache persist checkpoint

### Transformations ###

/**
          * Transformations转换算子
          * 延迟执行  需要Action算子触发才执行
          */
        //map 1:1 数据是一行
    //    lines.map(one=>{
    //      one+"#"
    //    }).foreach(println)
    
        //flatMap 1:n 数据是多行拆分过的
    //    lines.flatMap(one=>{one.split(" ")}).foreach(println)
    
        //过滤-filter
    //    val rdd1: RDD[String] = lines.flatMap(one=>{one.split(" ")})
    //    rdd1.filter(one=>{
    //      "book".equals(one)
    //    }).foreach(println)
    
        //排序-ortByKey
    //    val words: RDD[String] = lines.flatMap(one=>{one.split(" ")})
    //    val pairWords: RDD[(String, Int)] = words.map(one=>{(one,1)})
    //    val reduceResult: RDD[(String, Int)] = pairWords.reduceByKey((v1:Int,v2:Int)=>{v1+v2})
    //    //按照k(String)进行v的逆序
    //    //从map中取出数据并翻转，此时是数据在前单词在后
    //    val teransRDD: RDD[(Int, String)] = reduceResult.map(tp=>{tp.swap})
    //    val result: RDD[(Int, String)] = teransRDD.sortByKey(false)
    //    //再次转换kv位置
    //    result.map(_.swap).foreach(println)
    
        //排序-sortBy
        //按照v(Int)进行排序，不写false，默认正序排列
    //    reduceResult.sortBy(tp=>{tp._2},false)foreach(println)
    
        //随机抽样-sample
        //(抽到所需数据后是继续，抽样占总数的比率，抽样的随机数据会固定)
    //    val result: RDD[String] = lines.sample(true,0.1,100L)
    //    result.foreach(println)

### Action ###

/**
          * Action行动算子
          * 触发执行
          */
        //拿取数据的前5条
        val result: Array[String] = lines.take(3)
        result.foreach(println)
    
        //拿取数据第第一条
        val result1: String = lines.first()
        println(result1)
    
        //计数数据
        val l = lines.count()
        println(l)
    
        //将计算结果回收到Driver端，在大数据情况中不建议使用
    //    val result2: Array[String] = lines.collect()
    //    result2.foreach(println)

### 控制算子-持久化 ###

#### 控制算子 -cache ####

默认将RDD的数据持久化到内存中，cache是懒执行，需要触发算子触发

注意：chche () = persist()=persist(StorageLevel.Memory\_Only)

#### 控制算子 -persist ####

可以指定持久化的级别。最常用的是MEMORY\_ONLY和MEMORY\_AND\_DISK。”\_2”表示有副本数。

常用的级别：  
 MEMORY\_ONLY  
 MEMORY\_ONLY\_SER  
 MEMORY\_AND\_DISK  
 MEMORY\_AND\_DISK\_SER

#### 控制算子 -checkpoint ####

当RDD的lineage某些逻辑非常复杂，还可以使用checkpoint对RDD进行持久化  
不要经常对RDD使用checkpoint，这样会导致application执行缓慢，因为使用checkpoint相当于从磁盘中读取数据  
checkpoint执行流程  
 1.当spark执行完成之后，Spark框架会从后往前回溯，并找到chekpointRDD进行标记  
 2.回溯完成之后，Spark会重新启动Job，查看并计算chekpointRDDRDD的数据，将该RDD的数据存放在指定的持久化目录中  
 3.切断RDD之间的依赖关系  
 优化：持久化之前最好先cache下

## 三、Spark运行模式 ##

local： 多用于本地测试，如在eclipse，idea中写程序测试等。  
standalone: Standalone是Spark自带的一个资源调度框架，它支持完全分布式。  
yarn: Hadoop生态圈里面的一个资源调度框架，Spark也是可以基于Yarn来计算的。  
mesos: 资源调度框架。

## 四、Spark代码执行流程 ##

1 SparkConf  
2 SparkContext  
3 创建RDD  
4 对RDD使用transformation算子转换  
5 对RDD使用action算子触发执行  
6 f.sc.stop()

## 五、集群搭建 ##

1.修改slaves.template

添加从节点eg：node2 node3

[2.修改spark-env.sh][2._spark-env.sh]

SPARK\_MASTER\_IP:node1 master的ip

SPARK\_MASTER\_PORT:7077 提交任务的端口，默认是7077

SPARK\_WORKER\_CORES:2 每个worker从节点能够支配的core的个数

SPARK\_WORKER\_MEMORY:3g 每个worker从节点能够支配的内存数

3.主节点配置完成后分发到2,3节点

4.解压原包（不需要修改任何数据）到新节点node4，此节点用于提交任务