Spark学习笔记之Spark中的RDD的具体使用
程序员文章站
2022-06-02 10:32:36
1. spark中的rdd
resilient distributed datasets(弹性分布式数据集)
spark中的最基本的抽象
有了rdd的...
1. spark中的rdd
- resilient distributed datasets(弹性分布式数据集)
- spark中的最基本的抽象
- 有了rdd的存在我们就可以像操作本地集合一样操作分布式的数据
- 包含所有元素的分区的集合
- rdd包含了很多的分区
- 2. rdd中的弹性
- rdd中的数据是可大可小的
- rdd的数据默认情况下存放在内存中的,但是在内存资源不足时,spark会自动将rdd数据写入磁盘
- rdd有自动容错功能,当其中一个rdd中的分区的数据丢失,或者当前节点故障时,rdd会根据依赖关系重新计算该分区的数据
3. rdd在spark中的作用
迭代式计算
其主要实现思想就是rdd,把所有计算的数据保存在分布式的内存中。迭代计算通常情况下都是对同一个数据集做反复的迭代计算,数据在内存中将大大提升io操作。这也是spark涉及的核心:内存计算
交互式计算
因为spark是用scala语言实现的,spark和scala能够紧密的集成,所以spark可以完美的运用scala的解释器,使得其中的scala可以向操作本地集合对象一样轻松操作分布式数据集
4. spark中的名词解释
- clustermanager :在standalone模式中即为master(主节点),控制整个集群,监控worker。在yarn模式中为资源管理器
- worker:从节点,负责控制计算节点,启动executor。在yarn模式中为nodemanager,负责计算节点的控制。
- driver 运行application的main()函数并创建sparkcontext
- executor (coarsegrainedexecutorbackend)在worker node上执行任务的组件、用于启动线程池运行任务。每个application拥有独立的一组executors
- sparkcontext :整个应用的上下文,控制应用的生命周期
- rdd :spark中的最基本的数据抽象
- dag scheduler : 根据dag(有向无环图)切分stage,并且生成task,以taskset的形式返回
- task schedual: 调度task,把task交给executor
- stage: 一个spark作业一般包含一到多个stage。
- task :一个stage包含一到多个task,通过多个task实现并行运行的功能
- transformations :转换操作,transformation是lazy的,不会马上执行,只有当调用action时才会执行
- actions : 动作
- sparkenv : 线程级别的上下文,存储运行时的重要组件的引用
5. 创建rdd的两种方式
通过并行化集合创建rdd(用于测试)
val list = list("java c++ java","java java java c++") val rdd = sc.parallelize(list)
通过加载hdfs中的数据创建rdd(生产环境)
val rdd = sc.textfile("hdfs://uplooking01:8020/sparktest/")
6. idea开发spark
6.1 pom依赖
<?xml version="1.0" encoding="utf-8"?> <project xmlns="http://maven.apache.org/pom/4.0.0" xmlns:xsi="http://www.w3.org/2001/xmlschema-instance" xsi:schemalocation="http://maven.apache.org/pom/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd"> <modelversion>4.0.0</modelversion> <groupid>com.uplooking.bigdata</groupid> <artifactid>2018-11-08-spark</artifactid> <version>1.0-snapshot</version> <properties> <maven.compiler.source>1.8</maven.compiler.source> <maven.compiler.target>1.8</maven.compiler.target> <scala.version>2.11.8</scala.version> <spark.version>2.2.0</spark.version> <hadoop.version>2.7.5</hadoop.version> </properties> <dependencies> <!-- 导入scala的依赖 --> <dependency> <groupid>org.scala-lang</groupid> <artifactid>scala-library</artifactid> <version>${scala.version}</version> </dependency> <!-- 导入spark的依赖 --> <dependency> <groupid>org.apache.spark</groupid> <artifactid>spark-core_2.11</artifactid> <version>${spark.version}</version> </dependency> <dependency> <groupid>org.apache.spark</groupid> <artifactid>spark-sql_2.11</artifactid> <version>${spark.version}</version> </dependency> <!-- 指定hadoop-client api的版本 --> <dependency> <groupid>org.apache.hadoop</groupid> <artifactid>hadoop-client</artifactid> <version>${hadoop.version}</version> </dependency> </dependencies> <build> <plugins> <!--编译scala--> <plugin> <groupid>net.alchim31.maven</groupid> <artifactid>scala-maven-plugin</artifactid> <version>3.2.2</version> <executions> <execution> <id>scala-compile-first</id> <phase>process-resources</phase> <goals> <goal>add-source</goal> <goal>compile</goal> </goals> </execution> <execution> <id>scala-test-compile</id> <phase>process-test-resources</phase> <goals> <goal>testcompile</goal> </goals> </execution> </executions> </plugin> <!--编译java--> <plugin> <groupid>org.apache.maven.plugins</groupid> <artifactid>maven-compiler-plugin</artifactid> <executions> <execution> <phase>compile</phase> <goals> <goal>compile</goal> </goals> </execution> </executions> </plugin> <!-- 打jar插件 --> <plugin> <groupid>org.apache.maven.plugins</groupid> <artifactid>maven-shade-plugin</artifactid> <version>2.4.3</version> <configuration> <createdependencyreducedpom>false</createdependencyreducedpom> </configuration> <executions> <execution> <phase>package</phase> <goals> <goal>shade</goal> </goals> <configuration> <filters> <filter> <artifact>*:*</artifact> <excludes> <exclude>meta-inf/*.sf</exclude> <exclude>meta-inf/*.dsa</exclude> <exclude>meta-inf/*.rsa</exclude> </excludes> </filter> </filters> </configuration> </execution> </executions> </plugin> </plugins> </build> </project>
6.2 编写spark程序
val conf = new sparkconf() conf.setappname("ops1") val sc = new sparkcontext(conf) val rdd1: rdd[string] = sc.parallelize(list("java c+ java", "java java c++")) val ret = rdd1.collect().tobuffer println(ret)
6.3 打包
6.4 在driver上运行jar包
spark-submit --master spark://uplooking01:7077 --class com.uplooking.bigdata.spark01.ops1 original-spark-1.0-snapshot.jar
7. 本地运行spark程序
import org.apache.spark.rdd.rdd import org.apache.spark.{sparkconf, sparkcontext} import scala.collection.mutable object ops1 { def main(args: array[string]): unit = { val conf = new sparkconf() conf.setappname("ops1") conf.setmaster("local[4]") val sc = new sparkcontext(conf) //一般不会指定最小分区数 val rdd1 = sc.textfile("hdfs://uplooking01:8020/sparktest/") val rdd2: rdd[string] = rdd1.flatmap(line => line.split(" ")) val rdd3: rdd[(string, int)] = rdd2.map(word => (word, 1)) val rdd4: rdd[(string, int)] = rdd3.reducebykey(_ + _) val ret: mutable.buffer[(string, int)] = rdd4.collect().tobuffer println(ret) println(rdd1.partitions.length) } }
8. rdd中的分区数
并行化的方式指定分区数(一般会指定分区数)
- 默认如果创建rdd时不指定分区数,那么就会创建cpu核数个分区
- 手动指定分区数
val rdd = sc.parallelize(list("java c+ java", "java java c++"), 2)
textfile的方式指定分区数
- 默认如果创建rdd时不指定最小分区数,那么就会创建至少2个分区的rdd
- 一般不会指定最小分区数
- 不指定最小分区数,有切片的数量个分区
9. spark作业的运行流程
- 构建dag
- 根据dag切分stage,每个stage对应一组相同计算逻辑不能计算数据的task,以tastset的形式返回
- taskschedual调度task,把task发送到executor中去,用runnable进行包装进给线程池
- executor执行task
以上就是本文的全部内容,希望对大家的学习有所帮助,也希望大家多多支持。
上一篇: CDR简单制作漂亮的3D立体字
推荐阅读
-
ng2学习笔记之bootstrap中的component使用教程
-
Spark学习笔记之RDD中的Transformation和Action函数
-
Vue 2.0学习笔记之使用$refs访问Vue中的DOM
-
11.Spark Streaming源码解读之Driver中的ReceiverTracker架构设计以及具体实现彻底研究
-
11.Spark Streaming源码解读之Driver中的ReceiverTracker架构设计以及具体实现彻底研究
-
Spark学习笔记Spark Streaming的使用
-
Spark学习笔记之Spark SQL的具体使用
-
Spark学习笔记之Spark中的RDD的具体使用
-
Spark学习笔记之RDD中的Transformation和Action函数
-
Oracle学习笔记之Oracle中物化视图的使用