前言

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幾種部署方式？

Local(local[*],所有的cpu cores)
StandAlone(Master-local)
StandAloneHA(多個Master)
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 completed
  stream.asInstanceOf[CanCommitOffsets].commitAsync(offsetRanges)
}

package cn.it.sparkstreaming.kafka

import 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-呼叫streamingCOntext
    val 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進行累加求和wordcount
    val resultRDD: DStream[(String, Int)] = mapValue
      .flatMap(_.split("\\s+"))
      .map((_, 1))
      .updateStateByKey(updateFunc)
    resultRDD.print()
    //6-ssc.statrt
    ssc.start()
    //7-ssc.awaitTermination
    ssc.awaitTermination()
    //8-ssc.stop(true,true)
    ssc.stop(true, true)
  }
}

結構化流整合kafka

package cn.it.structedstreaming.kafka

import 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.awaitTermination
    query.awaitTermination()
    //6-query.stop
    query.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博客🙉
📢大資料系列文章會每天更新，停下休息的時候不要忘了別人還在奔跑，希望大家抓緊時間學習，全力奔赴更美好的生活?

轉載請註明出處，本文鏈接：https://www.uj5u.com/qita/289221.html

標籤：其他

上一篇：SparkStreaming的原理介紹

下一篇：2021年大資料HBase（十二）：Apache Phoenix 二級索引

??Spark的關鍵技識訓顧，持續更新！【推薦收藏加關注】??

前言

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來不及消費，請問造成什么現象？