Spark学习笔记之Spark中的RDD的具体使用
1.Spark中的RDD
- ResilientDistributedDatasets(弹性分布式数据集)
- 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)在workernode上执行任务的组件、用于启动线程池运行任务。每个Application拥有独立的一组Executors
- SparkContext:整个应用的上下文,控制应用的生命周期
- RDD:Spark中的最基本的数据抽象
- DAGScheduler:根据DAG(有向无环图)切分stage,并且生成task,以taskset的形式返回
- TaskSchedual:调度task,把task交给executor
- Stage:一个Spark作业一般包含一到多个Stage。
- Task:一个Stage包含一到多个Task,通过多个Task实现并行运行的功能
- Transformations:转换操作,Transformation是lazy的,不会马上执行,只有当调用action时才会执行
- Actions:动作
- SparkEnv:线程级别的上下文,存储运行时的重要组件的引用
5.创建RDD的两种方式
通过并行化集合创建RDD(用于测试)
vallist=List("javac++java","javajavajavac++") valrdd=sc.parallelize(list)
通过加载hdfs中的数据创建RDD(生产环境)
valrdd=sc.textFile("hdfs://uplooking01:8020/sparktest/")
6.IDEA开发Spark
6.1pom依赖
4.0.0 com.uplooking.bigdata 2018-11-08-spark 1.0-SNAPSHOT 1.8 1.8 2.11.8 2.2.0 2.7.5 org.scala-lang scala-library ${scala.version} org.apache.spark spark-core_2.11 ${spark.version} org.apache.spark spark-sql_2.11 ${spark.version} org.apache.hadoop hadoop-client ${hadoop.version} net.alchim31.maven scala-maven-plugin 3.2.2 scala-compile-first process-resources add-source compile scala-test-compile process-test-resources testCompile org.apache.maven.plugins maven-compiler-plugin compile compile org.apache.maven.plugins maven-shade-plugin 2.4.3 false package shade *:* META-INF/*.SF META-INF/*.DSA META-INF/*.RSA
6.2编写spark程序
valconf=newSparkConf() conf.setAppName("Ops1") valsc=newSparkContext(conf) valrdd1:RDD[String]=sc.parallelize(List("javac+java","javajavac++")) valret=rdd1.collect().toBuffer println(ret)
6.3打包
6.4在Driver上运行jar包
spark-submit--masterspark://uplooking01:7077--classcom.uplooking.bigdata.spark01.Ops1original-spark-1.0-SNAPSHOT.jar
7.本地运行Spark程序
importorg.apache.spark.rdd.RDD importorg.apache.spark.{SparkConf,SparkContext} importscala.collection.mutable objectOps1{ defmain(args:Array[String]):Unit={ valconf=newSparkConf() conf.setAppName("Ops1") conf.setMaster("local[4]") valsc=newSparkContext(conf) //一般不会指定最小分区数 valrdd1=sc.textFile("hdfs://uplooking01:8020/sparktest/") valrdd2:RDD[String]=rdd1.flatMap(line=>line.split("")) valrdd3:RDD[(String,Int)]=rdd2.map(word=>(word,1)) valrdd4:RDD[(String,Int)]=rdd3.reduceByKey(_+_) valret:mutable.Buffer[(String,Int)]=rdd4.collect().toBuffer println(ret) println(rdd1.partitions.length) } }
8.RDD中的分区数
并行化的方式指定分区数(一般会指定分区数)
- 默认如果创建RDD时不指定分区数,那么就会创建cpu核数个分区
- 手动指定分区数
valrdd=sc.parallelize(List("javac+java","javajavac++"),2)
textFile的方式指定分区数
- 默认如果创建RDD时不指定最小分区数,那么就会创建至少2个分区的RDD
- 一般不会指定最小分区数
- 不指定最小分区数,有切片的数量个分区
9.Spark作业的运行流程
- 构建DAG
- 根据DAG切分Stage,每个Stage对应一组相同计算逻辑不能计算数据的Task,以TastSet的形式返回
- TaskSchedual调度task,把task发送到executor中去,用Runnable进行包装进给线程池
- Executor执行task
以上就是本文的全部内容,希望对大家的学习有所帮助,也希望大家多多支持毛票票。