如何在Spark中使用RDD

如何在Spark中使用RDD？针对这个问题，这篇文章详细介绍了相对应的分析和解答，希望可以帮助更多想解决这个问题的小伙伴找到更简单易行的方法。

成都创新互联公司是一家以网站建设公司、网页设计、品牌设计、软件运维、成都网站推广、小程序App开发等移动开发为一体互联网公司。已累计为成都发电机租赁等众行业中小客户提供优质的互联网建站和软件开发服务。

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依赖

  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程序

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

关于如何在Spark中使用RDD问题的解答就分享到这里了，希望以上内容可以对大家有一定的帮助，如果你还有很多疑惑没有解开，可以关注创新互联行业资讯频道了解更多相关知识。