Spark学习笔记之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

以上就是本文的全部内容，希望对大家的学习有所帮助，也希望大家多多支持。