目錄

1. 簡(jiǎn)述Spark持久化中緩存和checkpoint檢查點(diǎn)的區(qū)別

2 . 如何使用緩存和檢查點(diǎn)?

3 . 代碼題

瀏覽器Nginx案例

先進(jìn)行數(shù)據(jù)清洗,做后續(xù)需求用

1、需求一：點(diǎn)擊最多的前10個(gè)網(wǎng)站域名

2、需求二：用戶最喜歡點(diǎn)擊的頁(yè)面排序TOP10

3、需求三：統(tǒng)計(jì)每分鐘用戶搜索次數(shù)

學(xué)生系統(tǒng)案例

4. RDD依賴的分類

5. 簡(jiǎn)述DAG與Stage 形成過(guò)程?

DAG :??

Stage :?

---

1. 簡(jiǎn)述Spark持久化中緩存和checkpoint檢查點(diǎn)的區(qū)別

1- 數(shù)據(jù)存儲(chǔ)位置不同
?? ?緩存: 存儲(chǔ)在內(nèi)存或者磁盤 或者 堆外內(nèi)存中
?? ?checkpoint檢查點(diǎn): 可以將數(shù)據(jù)存儲(chǔ)在磁盤或者HDFS上, 在集群模式下, 僅能保存到HDFS上

2- 數(shù)據(jù)生命周期:
?? ?緩存: 當(dāng)程序執(zhí)行完成后, 或者手動(dòng)調(diào)用unpersist 緩存都會(huì)被刪除
?? ?checkpoint檢查點(diǎn): 即使程序退出后, checkpoint檢查點(diǎn)的數(shù)據(jù)依然是存在的, 不會(huì)刪除, 需要手動(dòng)刪除

3- 血緣關(guān)系:
?? ?緩存: 不會(huì)截?cái)郣DD之間的血緣關(guān)系, 因?yàn)榫彺鏀?shù)據(jù)有可能是失效, 當(dāng)失效后, 需要重新回溯計(jì)算操作
?? ?checkpoint檢查點(diǎn): 會(huì)截?cái)嗟粢蕾囮P(guān)系, 因?yàn)閏heckpoint將數(shù)據(jù)保存到更加安全可靠的位置, 不會(huì)發(fā)生數(shù)據(jù)丟失的問(wèn)題, 當(dāng)執(zhí)行失敗的時(shí)候, 也不需要重新回溯執(zhí)行
?? ?
4- 主要作用不同：
?? ?緩存: 提高Spark程序的運(yùn)行效率
?? ?checkpoint檢查點(diǎn): 提高Spark程序的容錯(cuò)性

2 . 如何使用緩存和檢查點(diǎn)?

? ? ? ?將兩種方案同時(shí)用在一個(gè)項(xiàng)目中, 先設(shè)置緩存,再設(shè)置檢查點(diǎn) ,? 最后一同使用Action算子進(jìn)行觸發(fā), 這樣程序只會(huì)有一次IO操作, 如果先設(shè)置檢查點(diǎn)的話,就會(huì)有2次IO操作;

? ? ? ? ?當(dāng)在后續(xù)工程中讀取數(shù)據(jù)的時(shí)候,優(yōu)先從緩存中讀取,如果緩存中沒(méi)有數(shù)據(jù), 再?gòu)臋z查點(diǎn)讀取數(shù)據(jù),并且會(huì)將數(shù)據(jù)緩存一份到內(nèi)存中 ,后續(xù)直接從緩存中讀取數(shù)據(jù)

3 . 代碼題

瀏覽器Nginx案例

????????

先進(jìn)行數(shù)據(jù)清洗,做后續(xù)需求用

import os
from pyspark import SparkConf, SparkContext,StorageLevel
from pyspark.sql import SparkSession
import pyspark.sql.functions as F# 綁定指定的Python解釋器
os.environ['SPARK_HOME'] = '/export/server/spark'
os.environ['PYSPARK_PYTHON'] = '/root/anaconda3/bin/python3'
os.environ['PYSPARK_DRIVER_PYTHON'] = '/root/anaconda3/bin/python3'
# 綁定指定的Python解釋器
from pyspark.sql.types import StructType, IntegerType, StringType, StructFieldif __name__ == '__main__':
# 1- 創(chuàng)建SparkSession對(duì)象conf = SparkConf().setAppName('需求1').setMaster('local[*]')sc = SparkContext(conf=conf)
# 2- 數(shù)據(jù)輸入init_rdd = sc.textFile('file:///export/data/2024.1.2_Spark/1.6_day04/SogouQ.sample')# 3- 數(shù)據(jù)處理filter_tmp_rdd = init_rdd.filter(lambda line:line.strip()!='')print('過(guò)濾空行的數(shù)據(jù)',filter_tmp_rdd.take(10))map_rdd = filter_tmp_rdd.map(lambda line:line.split())print('map出來(lái)的數(shù)據(jù)',map_rdd.take(10))len6_rdd = map_rdd.filter(lambda line:len(line)==6)print('字段數(shù)為6個(gè)的字段',len6_rdd.take(10))etl_rdd = len6_rdd.map(lambda list:(list[0],list[1],list[2][1:-1],list[3],list[4],list[5])  )print('轉(zhuǎn)換成元組后的數(shù)據(jù)',etl_rdd.take(10))# 設(shè)置緩存etl_rdd.persist(storageLevel=StorageLevel.MEMORY_AND_DISK).count()

1、需求一：點(diǎn)擊最多的前10個(gè)網(wǎng)站域名

print('點(diǎn)擊最多的前10個(gè)網(wǎng)站域名','-'*50)website_map_rdd = etl_rdd.map(lambda tup:(tup[5].split('/')[0],1))print('把網(wǎng)站域名切出來(lái),變成(hello,1)的格式',website_map_rdd.take(10))website_reducekey_rdd = website_map_rdd.reduceByKey(lambda agg,curr:agg+curr)print('進(jìn)行聚合',website_reducekey_rdd.take(10))sort_rdd =website_reducekey_rdd.sortBy(lambda tup:tup[1],ascending=False)print('進(jìn)行降序排序',sort_rdd.take(10))
# 4- 數(shù)據(jù)輸出
# 5- 釋放資源sc.stop()

2、需求二：用戶最喜歡點(diǎn)擊的頁(yè)面排序TOP10

    print('用戶最喜歡點(diǎn)擊的頁(yè)面排序TOP10','-'*100)top_10_order = etl_rdd.map(lambda tup:(tup[4],1))print('點(diǎn)擊量排行',top_10_order.take(10))top_10_reducebykey = top_10_order.reduceByKey(lambda agg,curr:agg+curr)print('進(jìn)行聚合',top_10_reducebykey.take(10))sortby_top10 = top_10_reducebykey.sortBy(lambda line:line[1],ascending=False)print('進(jìn)行排序',sortby_top10.take(10))
# 4- 數(shù)據(jù)輸出
# 5- 釋放資源sc.stop()

3、需求三：統(tǒng)計(jì)每分鐘用戶搜索次數(shù)

    print('統(tǒng)計(jì)每分鐘用戶搜索次數(shù)','-'*50)search_map_rdd = etl_rdd.map(lambda tup:(tup[0][0:5],1))print('把網(wǎng)站域名切出來(lái),變成(hello,1)的格式',search_map_rdd.take(10))search_reducekey_rdd = search_map_rdd.reduceByKey(lambda agg,curr:agg+curr)print('進(jìn)行聚合',search_reducekey_rdd.take(10))sort_rdd =search_reducekey_rdd.sortBy(lambda tup:tup)print('按照時(shí)間進(jìn)行排序',sort_rdd.take(10))
# 4- 數(shù)據(jù)輸出
# 5- 釋放資源sc.stop()

學(xué)生系統(tǒng)案例

?數(shù)據(jù)準(zhǔn)備

import os
from pyspark import SparkConf, SparkContext, StorageLevel
from pyspark.sql import SparkSession
import pyspark.sql.functions as F# 綁定指定的Python解釋器
os.environ['SPARK_HOME'] = '/export/server/spark'
os.environ['PYSPARK_PYTHON'] = '/root/anaconda3/bin/python3'
os.environ['PYSPARK_DRIVER_PYTHON'] = '/root/anaconda3/bin/python3'
# 綁定指定的Python解釋器
from pyspark.sql.types import StructType, IntegerType, StringType, StructFieldif __name__ == '__main__':
# 1- 創(chuàng)建SparkSession對(duì)象conf = SparkConf().setAppName('學(xué)生案例').setMaster('local[*]')sc = SparkContext(conf=conf)
# 2- 數(shù)據(jù)輸入init_rdd= sc.textFile('hdfs://node1:8020/input/day04_home_work.txt')# 3- 數(shù)據(jù)處理stu_rdd = init_rdd.map(lambda line:line.split(',')).cache()print('切分后的數(shù)據(jù)為',stu_rdd.collect())

# 1、需求一：該系總共有多少學(xué)生stu_cnt = stu_rdd.map(lambda line:line[0]).distinct().count()print(f'該系總共有{stu_cnt}個(gè)學(xué)生')
# 2、需求二：該系共開設(shè)了多少門課程subject_cnt = stu_rdd.map(lambda line:line[1]).distinct().count()print(f'該系共開設(shè)了{(lán)subject_cnt}門課程')
# 3、需求三：Tom同學(xué)的總成績(jī)平均分是多少tom_score_sum = stu_rdd.filter(lambda line:line[0]=='Tom').map(lambda line:int(line[2])).sum()tom_subject_num = stu_rdd.filter(lambda line:line[0]=='Tom').map(lambda line:line[1]).distinct().count()tom_score_avg = tom_score_sum/tom_subject_numprint(f'Tom同學(xué)的總成績(jī)平均分是{round(tom_score_avg,2)}')# 4、需求四：求每名同學(xué)的選修的課程門數(shù)
#     every_student_course_num = stu_rdd.map(lambda x: (x[0], x[1])).distinct().map(lambda tup: (tup[0], 1))\
#         .reduceByKey(lambda agg, curr: agg + curr).collect()every_student_course_num = stu_rdd.map(lambda x: (x[0], x[1])).distinct()print('學(xué)生與選修課,把一個(gè)學(xué)生重修一門選修課的情況去掉',every_student_course_num.collect())every_student_course_num2 = every_student_course_num\.map(lambda tup:(tup[0],1))\.reduceByKey(lambda agg,curr:agg+curr).collect()print('每個(gè)同學(xué)的選修課數(shù)',every_student_course_num2)
# 5、需求五：該系DataBase課程共有多少人選修subject_database = stu_rdd.filter(lambda line:line[1]=='DataBase').map(lambda line:line[0]).distinct().count()print(f'數(shù)據(jù)庫(kù)有{subject_database}人選修')
# 6、需求六：各門課程的平均分是多少total_score = stu_rdd.map(lambda x:(x[1],int(x[2]))).groupByKey().map(lambda x:(x[0],sum(x[1])))print('各科總分為',total_score.collect())total_num = stu_rdd.map(lambda x: (x[1], 1)).groupByKey().map(lambda x: (x[0], sum(x[1])))print('各科的數(shù)量為',total_num.collect())#total_join =total_score.join(total_num)print('join后結(jié)果',total_join.collect())
# 各科總分為 [('DataBase', 170), ('Algorithm', 110), ('DataStructure', 140)]
# 各科的數(shù)量為 [('DataBase', 2), ('Algorithm', 2), ('DataStructure', 2)]
# 合表后為 [('DataBase', (170, 2)), ('DataStructure', (140, 2)), ('Algorithm', (110, 2))]total_avg =total_score.join(total_num).map(lambda x: (x[0], round(x[1][0] / x[1][1], 2))).collect()print('各科目的平均分為',total_avg)
# 4- 數(shù)據(jù)輸出# 5- 釋放資源sc.stop()

4. RDD依賴的分類

窄依賴:? 父RDD分區(qū)與子RDD分區(qū)是一對(duì)一關(guān)系

寬依賴:? 父RDD分區(qū)與子RDD分區(qū)是一對(duì)多關(guān)系

5. 簡(jiǎn)述DAG與Stage 形成過(guò)程?

DAG :??

1-Spark應(yīng)用程序，遇到了Action算子以后，就會(huì)觸發(fā)一個(gè)Job任務(wù)的產(chǎn)生。Job任務(wù)首先將它所依賴的全部算子加載到內(nèi)存中，形成一個(gè)完整Stage

2-會(huì)根據(jù)算子間的依賴關(guān)系，從Action算子開始，從后往前進(jìn)行回溯，如果算子間是窄依賴，就放到同一個(gè)Stage中；如果是寬依賴，就形成新的Stage。一直回溯完成。

Stage :?

1-Driver進(jìn)程啟動(dòng)成功以后，底層基于PY4J創(chuàng)建SparkContext對(duì)象，在創(chuàng)建SparkContext對(duì)象的過(guò)程中，還會(huì)同時(shí)創(chuàng)建DAGScheduler（DAG調(diào)度器）和TaskScheduler（Task調(diào)度器）
?? ?DAGScheduler: 對(duì)Job任務(wù)形成DAG有向無(wú)環(huán)圖和劃分Stage階段
?? ?TaskScheduler: 調(diào)度Task線程給到Executor進(jìn)程進(jìn)行執(zhí)行

2-Spark應(yīng)用程序遇到了一個(gè)Action算子以后，就會(huì)觸發(fā)一個(gè)Job任務(wù)的產(chǎn)生。SparkContext對(duì)象將Job任務(wù)提交DAG調(diào)度器，對(duì)Job形成DAG有向無(wú)環(huán)圖和劃分Stage階段。并且確定每個(gè)Stage階段需要有多少個(gè)Task線程，將這些Task線程放置在TaskSet集合中。再將TaskSet集合給到Task調(diào)度器。

3-Task調(diào)度器接收到DAG調(diào)度器傳遞過(guò)來(lái)的TaskSet集合以后，將Task線程分配給到具體的Executor進(jìn)行執(zhí)行，底層是基于調(diào)度隊(duì)列SchedulerBackend。Stage階段是一個(gè)一個(gè)按順序執(zhí)行的，不能并行執(zhí)行。

4-Executor進(jìn)程開始執(zhí)行具體的Task線程。后續(xù)過(guò)程就是Driver監(jiān)控多個(gè)Executor的執(zhí)行狀態(tài)，直到Job任務(wù)執(zhí)行完成。