共计 5023 个字符,预计需要花费 13 分钟才能阅读完成。
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
正文完