目錄

前言

Spark的關鍵技術回顧

一、Spark復習題回顧

1、Spark使用的版本

2、Spark幾種部署方式？

3、Spark的提交任務的方式？

4、使用Spark-shell的方式也可以交互式寫Spark代碼？

5、你對RDD是怎么理解的？

6、Spark如何實現容錯？

7、Spark共享變量？

8、Spark的任務執行?

9、Spark的RDD的幾種類型？

10、Spark的Transformation算子有幾類？

11、RDD創建的三種方法？

12、RDD-DataSet和DataFrame的區別和聯系？

13、SparkSQL中查詢一列的字段的方法有幾種？

14、SparkSQL中的如何動態增加Schema??

15、SparkSQL中DSL和SQL風格差異？

16、SparkSQL中SQL風格全局Session和局部的Session的差別是什么？

17、SparkSQL整合Hive？

18、[非常重要]SparkSQL如何執行SQL的，SQL的查詢引擎

19、SparkStreaming幾種編程模式？

20、對于DStream如何使用RDD的方法?

21、SparkStreaming的有狀態的幾種形式？

22、SparkStreaming和Kafka的整合，如何獲取Offset，010整合

23、SparkStreaming有兩個時間？

24、生產者生產數據過多，消費者SparkStreaming來不及消費，請問造成什么現象？

---

前言

?2021大數據領域優質創作博客，帶你從入門到精通，該博客每天更新，逐漸完善大數據各個知識體系的文章，幫助大家更高效學習。

?

Spark的關鍵技術回顧

一、Spark復習題回顧

1、Spark使用的版本

2.4.5版本，目前3.1.2為最新版本

集群環境：CDH版本是5.14.0這個版本
但由于spark對應的5.14.0的CDH版本的軟件默認的版本是1.6.0同時閹割了SarkSQL，需要重新編譯
原因： 因為Cloudera公司認為有了impala就不需要再使用sparkSQL的功能了，同時也是為了推廣impala，所以直接閹割掉了sparkSQL的模塊。
解決： 使用Apache的版本的spark來進行重新編譯

2、Spark幾種部署方式？

1. Local(local[*],所有的cpu cores)
2. StandAlone(Master-local)
3. StandAloneHA(多個Master)
4. Yarn(RS-NM) --client 模式和cluster 模式

3、Spark的提交任務的方式？

bin/spark-submit ?\
--master local/spark:node01:7077/spark:node01:7077,node02:70777 \
--deploy-mode client/cluster \ ?#client指的是driver啟動在本地，cluster指的是driver啟動在Worker接點運行
--class application-main
--executor-memory?? ? 每個executor的內存，默認是1G
--total-executor-cores?? ? 所有executor總共的核數。僅僅在 mesos 或者 standalone 下使用
--executor-core?? ? 每個executor的核數。在yarn或者standalone下使用
--driver-memory?? ? Driver內存，默認 1G
--driver-cores?? ? Driver 的核數，默認是1。在 yarn 或者 standalone 下使用
--num-executors?? ? 啟動的executor數量。默認為2。在 yarn 下使用
.....
jar包地址
參數1 參數2

4、使用Spark-shell的方式也可以交互式寫Spark代碼？

bin/spark-shell --master local --executor-core 2 --executor-memory 512m

?

5、你對RDD是怎么理解的？

（1）RDD是彈性分布式數據集

（2）RDD有五大屬性：1-RDD是可分區的(0-1-2號分區) 2-RDD有作用函數(map) 3-RDD是依賴關系 4-對key-value的類型RDD的默認分區HashPartitoner 5-位置優先性
wordount的時候：
sc.textFile().flatmap().map().redyceByKey()
如何查看當前算子是什么分區器？函數rdd.partitioner

（3）RDD的寬依賴和窄依賴：根據父RDD有一個或多個子RDD對應，因為窄依賴可以在任務間并行，寬依賴會發生Shuffle，并不是所有的bykey算子都會產生shuffle？需要注意的是（1）分區器一致（2）分區個數一致

（4）RDD血緣關系linage：linage會記錄當前RDD依賴于上一個RDD，如果一個RDD失效可以重建RDD，容錯關鍵

（5）RDD的緩存：cache和persist，cache會將數據緩存在內存中，persist可以指定多種存儲級別，cache底層調用的是persist

（6）RDD的檢查點機制：Checkpoint會截斷所有的血緣關系，而緩存會將血緣的關系全部保存在內存或磁盤中

6、Spark如何實現容錯？

Spark會首先查看內存中是否已經cache或persist還原，否則查看linage是否checkpoint在hdfs中
根據依賴關系重建RDD

7、Spark共享變量？

累加器

Spark提供的Accumulator，主要用于多個節點對一個變量進行共享性的操作。Accumulator只提供了累加的功能，即確提供了多個task對一個變量并行操作的功能。但是task只能對Accumulator進行累加操作，不能讀取Accumulator的值，只有Driver程序可以讀取Accumulator的值

(在driver端定義的變量在executor端拿到的是副本，exector執行完計算不會更新到driver)

廣播變量

廣播變量允許開發人員在每個節點（Worker or Executor）緩存只讀變量，而不是在Task之間傳遞這些變量。使用廣播變量能夠高效地在集群每個節點創建大數據集的副本。同時Spark還使用高效的廣播算法分發這些變量，從而減少通信的開銷

(對于1M的數據，開啟1000個maptask，當前的1M的數據會發送到所有的task中進行計算，會產生1G網絡數據傳輸，引入廣播變量將1M數據共享在Executor中而不是task中，task共享的是一個變量的副本，廣播變量是只讀的，不能再exectour端修改)

8、Spark的任務執行?

1-Spark一個Application擁有多個job，一個action操作會出發一個Job劃分
2-Spark一個Job有多個Stages，發生shuffle操作觸發一個Stage的劃分
3-一個Stage有很多個tasksets，一個RDD的不同的分區就是代表的taskset，很多的taskset組成tasksets
4-一個taskset由很多個RDD的分區組成，一個RDD的分區的數據需要由一個task線程拉取執行，而不是進程

9、Spark的RDD的幾種類型？

transformation和action類型

1)Transformation轉換操作：返回一個新的RDD

所有Transformation函數都是Lazy，不會立即執行，需要Action函數觸發

2)Action動作操作：返回值不是RDD(無返回值或返回其他的)

所有Action函數立即執行（Eager），比如count、first、collect、take等

10、Spark的Transformation算子有幾類？

3類
單value：如mapValue，map，filter
雙value：union，zip，distinct
key-value類型：reduceBykey(一定不屬于Action算子)，foldByKey

轉換	含義
map(func)	返回一個新的RDD，該RDD由每一個輸入元素經過func函數轉換后組成
filter(func)	返回一個新的RDD，該RDD由經過func函數計算后返回值為true的輸入元素組成
flatMap(func)	類似于map，但是每一個輸入元素可以被映射為0或多個輸出元素(所以func應該返回一個序列，而不是單一元素)
mapPartitions(func)	類似于map，但獨立地在RDD的每一個分片上運行，因此在類型為T的RDD上運行時，func的函數類型必須是Iterator[T] => Iterator[U]
mapPartitionsWithIndex(func)	類似于mapPartitions，但func帶有一個整數參數表示分片的索引值，因此在類型為T的RDD上運行時，func的函數類型必須是 (Int, Interator[T]) => Iterator[U]
sample(withReplacement, fraction, seed)	根據fraction指定的比例對數據進行采樣，可以選擇是否使用隨機數進行替換，seed用于指定隨機數生成器種子
union(otherDataset)	對源RDD和參數RDD求并集后返回一個新的RDD
intersection(otherDataset)	對源RDD和參數RDD求交集后返回一個新的RDD
distinct([numTasks]))	對源RDD進行去重后返回一個新的RDD
groupByKey([numTasks]) ??	在一個(K,V)的RDD上調用，返回一個(K, Iterator[V])的RDD
reduceByKey(func, [numTasks])	在一個(K,V)的RDD上調用，返回一個(K,V)的RDD，使用指定的reduce函數，將相同key的值聚合到一起，與groupByKey類似，reduce任務的個數可以通過第二個可選的參數來設置
aggregateByKey(zeroValue)(seqOp, combOp, [numTasks])
sortByKey([ascending], [numTasks])	在一個(K,V)的RDD上調用，K必須實現Ordered接口，返回一個按照key進行排序的(K,V)的RDD
sortBy(func,[ascending], [numTasks])	與sortByKey類似，但是更靈活
join(otherDataset, [numTasks])	在類型為(K,V)和(K,W)的RDD上調用，返回一個相同key對應的所有元素對在一起的(K,(V,W))的RDD
cogroup(otherDataset, [numTasks])	在類型為(K,V)和(K,W)的RDD上調用，返回一個(K,(Iterable<V>,Iterable<W>))類型的RDD
cartesian(otherDataset)	笛卡爾積
pipe(command, [envVars])	對rdd進行管道操作
coalesce(numPartitions) ?	減少 RDD 的分區數到指定值。在過濾大量數據之后，可以執行此操作
repartition(numPartitions)	重新給 RDD 分區

?

11、RDD創建的三種方法？

sc.textfile,sc.makerdd/paralleise,RDD之間的轉換

12、RDD-DataSet和DataFrame的區別和聯系？

RDD+Scheme=DataFrame.as[]+泛型=DataSet.rdd=RDD，
DataFrame是弱類型的數據類型，在運行時候數據類型檢查，
DataSet是強類型的數據類型，在編譯時候進行類型檢查

?

13、SparkSQL中查詢一列的字段的方法有幾種？

df.select(['id']),
df.select(col('id')),
df.select(colomns('id')),
df.select('id),
df.select($"")

14、SparkSQL中的如何動態增加Schema??

查看DataFrame中Schema是什么，執行如下命令：

df.schema

Schema信息封裝在StructType中，包含很多StructField對象，源碼。

StructType 定義，是一個樣例類，屬性為StructField的數組

StructField 定義，同樣是一個樣例類，有四個屬性，其中字段名稱和類型為必填‘’

自定義Schema結構，官方提供的示例代碼：

StructedType(StructedFileld(data,name,nullable)::Nil),
new StructedType().add(data,name,nullable).add()

?

15、SparkSQL中DSL和SQL風格差異？

?DSL風格df.select,SQL風格需要注冊一張臨時表或試圖進行展示

基于DSL分析

調用DataFrame/Dataset中API（函數）分析數據，其中函數包含RDD中轉換函數和類似SQL語句函數，部分截圖如下：

基于SQL分析

將Dataset/DataFrame注冊為臨時視圖，編寫SQL執行分析，分為兩個步驟：

第一步、注冊為臨時視圖

?

第二步、編寫SQL，執行分析

?

?

16、SparkSQL中SQL風格全局Session和局部的Session的差別是什么？

全局的Session可以跨Session訪問注冊的臨時試圖或表，局部Session只能訪問臨時試圖或表

17、SparkSQL整合Hive？

SparkSQL除了引用Hive的元數據的信息之外，其他的Hive部分都沒有耦合
Spark引擎替代了HIve的執行引擎，可以在SPark程序中使用HIve的語法完成SQ的分析
第一步：將hive-site.xml拷貝到spark安裝路徑conf目錄
第二步：將mysql的連接驅動包拷貝到spark的jars目錄下
第三步：Hive開啟MetaStore服務
第四步：測試Sparksql整合Hive是否成功

18、[非常重要]SparkSQL如何執行SQL的，SQL的查詢引擎

基于規則優化（Rule-based optimization, RBO----過濾下推，常量折疊）-邏輯執行計劃中，進行邏輯計劃優化
基于代價優化（Cost-based optimization, CBO）----物理執行計劃中選擇最優物理執行計劃

?

?

19、SparkStreaming幾種編程模式？

有狀態(updateStateByKey\mapState)、無狀態(reduceByKey)、窗口操作(windows，reduceByKeyANdWIndows)

20、對于DStream如何使用RDD的方法?

(transform)

21、SparkStreaming的有狀態的幾種形式？

updateStateByKey\mapState

22、SparkStreaming和Kafka的整合，如何獲取Offset，010整合

KafkaUtils.createdirctstream(SSC,Kafka的parititon和Spark的eceutor是否在一個節點，Consumer.subscribe(Array(kafkatopic),params))
獲取Offset：StreamData.asInstanceOf[HasOffSetRanges].offsetRanges
提交Offset：StreamData.asInstanceOf[CancommitOffSetRanges].async(offSetRanges)
#http://spark.apache.org/docs/latest/streaming-kafka-0-10-integration.html

val kafkaParams = Map[String, Object]("bootstrap.servers" -> "localhost:9092,anotherhost:9092","key.deserializer" -> classOf[StringDeserializer],"value.deserializer" -> classOf[StringDeserializer],"group.id" -> "use_a_separate_group_id_for_each_stream","auto.offset.reset" -> "latest","enable.auto.commit" -> (false: java.lang.Boolean)
)val topics = Array("topicA", "topicB")
val stream = KafkaUtils.createDirectStream[String, String](streamingContext,PreferConsistent,Subscribe[String, String](topics, kafkaParams)
)
stream.map(record => (record.key, record.value))
stream.foreachRDD { rdd =>val offsetRanges = rdd.asInstanceOf[HasOffsetRanges].offsetRanges// some time later, after outputs have completedstream.asInstanceOf[CanCommitOffsets].commitAsync(offsetRanges)
}

package cn.it.sparkstreaming.kafkaimport org.apache.kafka.clients.consumer.ConsumerRecord
import org.apache.kafka.common.serialization.StringDeserializer
import org.apache.spark.SparkConf
import org.apache.spark.streaming.dstream.{DStream, InputDStream}
import org.apache.spark.streaming.kafka010.{ConsumerStrategies, KafkaUtils, LocationStrategies}
import org.apache.spark.streaming.{Seconds, StreamingContext}/*** DESC:* 1-導入有kafka和spark整合的Jar包* 2-調用streamingCOntext* 3-KafkaUtils.creatDriectlyStream的方法直接連接Kafka集群的分區* 4-獲取record記錄中的value的值* 5-根據value進行累加求和wordcount* 6-ssc.statrt* 7-ssc.awaitTermination* 8-ssc.stop(true,true)*/
object _01SparkStreamingKafkaAuto {def updateFunc(curentValue: Seq[Int], histouryValue: Option[Int]): Option[Int] = {val sum: Int = curentValue.sum + histouryValue.getOrElse(0)Option(sum)}val kafkaParams = Map[String, Object]("bootstrap.servers" -> "node1:9092","key.deserializer" -> classOf[StringDeserializer],"value.deserializer" -> classOf[StringDeserializer],"group.id" -> "spark_group",//offset的偏移量自動設置為最新偏移量，有幾種設置偏移量的方法// //這里的auto.offset.reset代表的是自動重置offset為latest就表示的是最新的偏移量，如果沒有偏移從最新的位置開始"auto.offset.reset" -> "latest",//是否自動提交，這里設置為自動提交，提交到kafka指導的__consumertopic中，有kafka自己維護，如果設置為false可以使用ckeckpoint或者是將offset存入mysql// //這里如果是false手動提交，默認由SparkStreaming提交到checkpoint中，在這里也可以根據用戶或程序員將offset偏移量提交到mysql或redis中"enable.auto.commit" -> (true: java.lang.Boolean),//自動設置提交的時間"auto.commit.interval.ms" -> "1000")def main(args: Array[String]): Unit = {//1-導入有kafka和spark整合的Jar包//2-調用streamingCOntextval ssc: StreamingContext = {val conf: SparkConf = new SparkConf().setAppName(this.getClass.getSimpleName.stripSuffix("$")).setMaster("local[*]")val ssc = new StreamingContext(conf, Seconds(5))ssc}ssc.checkpoint("data/baseoutput/cck3")//3-KafkaUtils.creatDriectlyStream的方法直接連接Kafka集群的分區//ssc: StreamingContext,//locationStrategy: LocationStrategy,//consumerStrategy: ConsumerStrategy[K, V]val streamRDD: InputDStream[ConsumerRecord[String, String]] = KafkaUtils.createDirectStream[String, String](ssc,LocationStrategies.PreferConsistent,ConsumerStrategies.Subscribe[String, String](Array("spark_kafka"), kafkaParams))//4-獲取record記錄中的value的值val mapValue: DStream[String] = streamRDD.map(_.value())//5-根據value進行累加求和wordcountval resultRDD: DStream[(String, Int)] = mapValue.flatMap(_.split("\\s+")).map((_, 1)).updateStateByKey(updateFunc)resultRDD.print()//6-ssc.statrtssc.start()//7-ssc.awaitTerminationssc.awaitTermination()//8-ssc.stop(true,true)ssc.stop(true, true)}
}

結構化流整合kafka

package cn.it.structedstreaming.kafkaimport org.apache.spark.SparkConf
import org.apache.spark.sql.streaming.{OutputMode, StreamingQuery, Trigger}
import org.apache.spark.sql.{DataFrame, Dataset, Row, SparkSession}/*** DESC:* * 1-準備上下文環境* * 2-讀取Kafka的數據* * 3-將Kafka的數據轉化，實現單詞統計技術* * 4-將得到結果寫入控制臺* * 5.query.awaitTermination* * 6-query.stop*/
object _01KafkaSourceWordcount {def main(args: Array[String]): Unit = {//1-準備上下文環境val conf: SparkConf = new SparkConf().setAppName(this.getClass.getSimpleName.stripSuffix("$")).setMaster("local[*]")val spark: SparkSession = SparkSession.builder().config(conf).config("spark.sql.shuffle.partitions", "4").getOrCreate()//spark.sparkContext.setLogLevel("WARN")import spark.implicits._//2-讀取Kafka的數據val streamDF: DataFrame = spark.readStream.format("kafka").option("kafka.bootstrap.servers", "node1:9092").option("subscribe", "wordstopic").load()//streamDF.printSchema()//root// |-- key: binary (nullable = true)// |-- value: binary (nullable = true)// |-- topic: string (nullable = true)// |-- partition: integer (nullable = true)// |-- offset: long (nullable = true)// |-- timestamp: timestamp (nullable = true)// |-- timestampType: integer (nullable = true)//3-將Kafka的數據轉化，實現單詞統計技術val result: Dataset[Row] = streamDF.selectExpr("cast (value as string)") //因為kafka得到的數據是binary類型的數據需要使用cast轉換.as[String].flatMap(x => x.split("\\s+")) // |-- value: string (nullable = true).groupBy($"value").count().orderBy('count.desc)//.groupBy("value")//4-將得到結果寫入控制臺val query: StreamingQuery = result.writeStream.format("console").outputMode(OutputMode.Complete()).trigger(Trigger.ProcessingTime(0)).option("numRows", 10).option("truncate", false).start()//5.query.awaitTerminationquery.awaitTermination()//6-query.stopquery.stop()}
}

23、SparkStreaming有兩個時間？

Spark Streaming接收器接收到的數據在存儲到Spark中之前的時間間隔被分成數據塊。 最低建議-50毫秒。
一個時間是接收器接受數據的時間--默認是200ms，數據到來每隔200ms獲取一次數據，合并數據形成DStream
一個時間是SParkStreaming獲取到數據后處理時間--StreamingContext(sc,Second(5))，這才是SparkStreaming批處理時間

24、生產者生產數據過多，消費者SparkStreaming來不及消費，請問造成什么現象？

背壓，或反壓

SparkStreaming反壓

在SParkStreaming中是默認關閉，在Flink中是默認開啟的，背壓在SParkStreaing中自動動態的根據接收器接受最大速率和kafka的topic的分區的個數確定

---

• 📢博客主頁：https://lansonli.blog.csdn.net
• 📢歡迎點贊 👍 收藏 ?留言 📝 如有錯誤敬請指正！
• 📢本文由 Lansonli 原創，首發于 CSDN博客🙉
• 📢大數據系列文章會每天更新，停下休息的時候不要忘了別人還在奔跑，希望大家抓緊時間學習，全力奔赴更美好的生活?

總結

以上是生活随笔為你收集整理的❤️Spark的关键技术回顾，持续更新！【推荐收藏加关注】❤️的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。