大數據系列——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涉及的核心：內存計算java
- 交互式計算
由於Spark是用scala語言實現的，Spark和scala可以緊密的集成，因此Spark能夠完美的運用scala的解釋器，使得其中的scala能夠向操做本地集合對象同樣輕鬆操做分佈式數據集node
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(用於測試)c++
val list = List("java c++ java","java java java c++") val rdd = sc.parallelize(list) -
經過加載hdfs中的數據建立RDD(生產環境)sql
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中的分區數
-
並行化的方式指定分區數(通常會指定分區數)apache
- 默認若是建立RDD時不指定分區數,那麼就會建立cpu核數個分區
-
手動指定分區數maven
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
相關文章
- 1. 大數據系列：Spark學習筆記
- 2. 大數據系列——Spark學習筆記之初識Spark
- 3. 大數據系列——Spark學習筆記之 Spark SQL
- 4. 【大數據學習】之 Spark-RDD core4
- 5. 大數據系列——Spark學習筆記Spark Streaming
- 6. Spark RDD學習筆記
- 7. Spark RDD概念學習系列之RDD的5大特點（五）
- 8. spark筆記之RDD的依賴關係
- 9. Spark系列文章（四）:Spark之RDD
- 10. 大數據學習筆記之Spark（六）：Spark內核解析
- 更多相關文章...
- • 您已經學習了 XML Schema，下一步學習什麼呢？ - XML Schema 教程
- • 我們已經學習了 SQL，下一步學習什麼呢？ - SQL 教程
- • Tomcat學習筆記（史上最全tomcat學習筆記）
- • 適用於PHP初學者的學習線路和建議
相關標籤/搜索
每日一句
-
每一个你不满意的现在，都有一个你没有努力的曾经。
歡迎關注本站公眾號,獲取更多信息
