2025年P(guān)ython大數(shù)據(jù)處理試卷Spark實(shí)戰(zhàn)版考試時(shí)間：______分鐘總分：______分姓名：______一、選擇題（每題2分，共20分）1.在Spark中，下列哪個(gè)操作屬于轉(zhuǎn)換操作（Transformation）？A.`collect()`B.`map()`C.`count()`D.`take()`2.SparkSQL中，用于執(zhí)行SQL語(yǔ)句并返回DataFrame的接口是？A.`SparkSession.builder()`B.`SparkContext()`C.`SQLContext()`D.`DataFrameReader`3.當(dāng)Spark作業(yè)出現(xiàn)內(nèi)存不足錯(cuò)誤時(shí)，除了增加集群內(nèi)存外，以下哪種方法是提高內(nèi)存利用率的常用手段？A.減少分區(qū)數(shù)B.使用更復(fù)雜的數(shù)據(jù)類型C.適當(dāng)?shù)厥褂胉persist()`或`cache()`緩存中間DataFrameD.增加shuffle操作4.下列關(guān)于SparkRDD的描述，錯(cuò)誤的是？A.RDD是Spark中彈性分布式數(shù)據(jù)集的簡(jiǎn)稱B.RDD提供了豐富的容錯(cuò)機(jī)制C.RDD的操作是惰性求值的D.RDD一旦創(chuàng)建，其數(shù)據(jù)就會(huì)被立即加載到內(nèi)存中5.在Spark中，如果一個(gè)操作會(huì)觸發(fā)數(shù)據(jù)的跨節(jié)點(diǎn)傳輸，該操作通常被稱為？A.轉(zhuǎn)換操作B.行動(dòng)操作C.緩存操作D.并行操作6.以下哪個(gè)組件是SparkStreaming用于處理實(shí)時(shí)數(shù)據(jù)的核心抽象？A.RDDB.DataFrameC.DStreamD.SparkSession7.當(dāng)使用Spark的DataFrameAPI進(jìn)行復(fù)雜查詢時(shí)，哪個(gè)操作通常用于連接兩個(gè)DataFrame？A.`filter()`B.`groupBy()`C.`join()`D.`union()`8.在Spark中，`repartition(numPartitions)`和`coalesce(numPartitions)`的主要區(qū)別在于？A.`repartition`會(huì)復(fù)制數(shù)據(jù)，而`coalesce`不會(huì)B.`repartition`適用于增加分區(qū)數(shù)，`coalesce`適用于減少分區(qū)數(shù)C.`repartition`是懶執(zhí)行的，`coalesce`是立即執(zhí)行的D.它們對(duì)性能的影響相同9.以下哪個(gè)Spark配置參數(shù)用于控制RDD的默認(rèn)分區(qū)數(shù)？A.`spark.executor.memory`B.`spark.default.parallelism`C.`spark.driver.memory`D.`spark.sql.shuffle.partitions`10.PySpark中，要從HDFS讀取一個(gè)文本文件并創(chuàng)建一個(gè)RDD，通常使用的代碼片段是？A.`spark.read.json("hdfs://path")`B.`sc.textFile("hdfs://path")`C.`spark.read.parquet("hdfs://path")`D.`sc.parallelize(list("hdfs://path"))`二、填空題（每空2分，共20分）1.Spark中的RDD主要通過(guò)______和______兩種方式創(chuàng)建。2.在SparkSQL中，`groupBy("columnName")`后通常需要跟一個(gè)______操作來(lái)執(zhí)行聚合。3.為了提高Spark作業(yè)的容錯(cuò)性，Spark會(huì)對(duì)RDD的數(shù)據(jù)進(jìn)行______。4.PySpark中，`spark.conf.get("key")`用于獲取______的配置信息。5.SparkStreaming處理數(shù)據(jù)時(shí)，一個(gè)持續(xù)的數(shù)據(jù)流被視為一個(gè)______對(duì)象。6.當(dāng)對(duì)DataFrame進(jìn)行排序時(shí)，使用______函數(shù)指定排序的列和順序。7.在Spark中，避免不必要的______是優(yōu)化Spark作業(yè)性能的關(guān)鍵策略之一。8.Spark作業(yè)的執(zhí)行過(guò)程主要分為______、調(diào)度和執(zhí)行三個(gè)階段。9.PySpark中，`df.write.saveAsTextFile("hdfs://path")`會(huì)將DataFrame數(shù)據(jù)保存為______格式。10.如果一個(gè)RDD的父RDD被丟棄，但該RDD仍然被其他RDD依賴且未被行動(dòng)操作觸發(fā)，則Spark會(huì)通過(guò)______機(jī)制來(lái)重新計(jì)算該RDD。三、簡(jiǎn)答題（每題5分，共15分）1.簡(jiǎn)述Spark的惰性求值機(jī)制及其優(yōu)點(diǎn)。2.解釋Spark中“數(shù)據(jù)傾斜”的概念，并簡(jiǎn)述一種常見的解決方法。3.比較SparkRDDAPI和DataFrame/DatasetAPI的優(yōu)缺點(diǎn)。四、編程題（共45分）1.（15分）編寫PySpark代碼，實(shí)現(xiàn)以下功能：*從HDFS上的`/user/data/input`路徑讀取一個(gè)名為`log.txt`的文本文件，創(chuàng)建一個(gè)RDD。*對(duì)RDD中的每一行文本進(jìn)行分詞（按空格分隔），創(chuàng)建一個(gè)新的RDD。*對(duì)分詞后的RDD中所有單詞的出現(xiàn)次數(shù)進(jìn)行統(tǒng)計(jì)，并按出現(xiàn)次數(shù)降序排序，取前10個(gè)單詞及其計(jì)數(shù)。*將結(jié)果輸出到HDFS上的`/user/data/output`路徑，格式為每行一個(gè)“單詞:計(jì)數(shù)”。2.（30分）假設(shè)有一個(gè)DataFrame`df`，包含以下三列：`user_id`(用戶ID,整數(shù)類型),`product_id`(產(chǎn)品ID,整數(shù)類型),`score`(評(píng)分,浮點(diǎn)類型)。請(qǐng)使用SparkSQL的DataFrameAPI完成以下任務(wù)：*計(jì)算每個(gè)用戶的平均評(píng)分，并創(chuàng)建一個(gè)新的DataFrame`user_avg_score_df`，包含`user_id`和`avg_score`兩列。*從`user_avg_score_df`中篩選出平均評(píng)分大于等于4.0的用戶。*計(jì)算這些高評(píng)分用戶的總評(píng)分人數(shù)。*計(jì)算平均評(píng)分最高的產(chǎn)品ID及其對(duì)應(yīng)的平均評(píng)分。*最后，將篩選出的高評(píng)分用戶DataFrame(`high_score_users_df`)和平均評(píng)分最高的產(chǎn)品信息DataFrame(`top_product_df`)合并，合并條件為`user_id`等于`product_id`，只保留兩個(gè)DataFrame都存在的記錄。將最終結(jié)果DataFrame輸出到HDFS上的`/user/data/report`路徑，格式為Parquet。---試卷答案一、選擇題1.B解析：`map()`是轉(zhuǎn)換操作，它對(duì)RDD中的每個(gè)元素應(yīng)用一個(gè)函數(shù)，產(chǎn)生一個(gè)新的RDD。`collect()`,`count()`,`take()`都是行動(dòng)操作，會(huì)觸發(fā)RDD的執(zhí)行并返回結(jié)果。2.C解析：`SQLContext`是SparkSQL早期版本中用于創(chuàng)建DataFrame和執(zhí)行SQL查詢的接口。在較新的Spark版本中，通常使用`SparkSession`，它是`SQLContext`的擴(kuò)展，提供了更全面的功能。但題目問(wèn)的是執(zhí)行SQL語(yǔ)句返回DataFrame的接口，`SQLContext`直接關(guān)聯(lián)了SQL功能。注意：最新版Spark推薦使用SparkSession。3.C解析：增加內(nèi)存是直接方法，但`persist()`或`cache()`通過(guò)將中間結(jié)果緩存到內(nèi)存（或磁盤中），減少重復(fù)計(jì)算，能有效提高內(nèi)存利用率和處理速度。減少分區(qū)數(shù)可能導(dǎo)致計(jì)算能力下降，使用復(fù)雜數(shù)據(jù)類型會(huì)增加內(nèi)存消耗，增加shuffle操作通常會(huì)加劇內(nèi)存和網(wǎng)絡(luò)壓力。4.D解析：RDD的數(shù)據(jù)默認(rèn)不會(huì)立即加載到內(nèi)存，Spark采用懶加載機(jī)制。數(shù)據(jù)只在需要時(shí)（觸發(fā)行動(dòng)操作）才被讀取和計(jì)算。RDD提供了容錯(cuò)機(jī)制（通過(guò)記錄依賴關(guān)系和重計(jì)算），操作是惰性求值的。5.B解析：行動(dòng)操作（ActionOperation）是RDDAPI中觸發(fā)實(shí)際計(jì)算和數(shù)據(jù)序列化操作的步驟，例如`collect()`,`count()`,`saveAsTextFile()`等。行動(dòng)操作會(huì)從RDD中讀取數(shù)據(jù)，并將其發(fā)送到驅(qū)動(dòng)程序或存儲(chǔ)系統(tǒng)，這是區(qū)別于轉(zhuǎn)換操作（TransformationOperation，如`map()`,`filter()`等，只定義數(shù)據(jù)轉(zhuǎn)換邏輯，不觸發(fā)計(jì)算）的關(guān)鍵。6.C解析：DStream（DiscretizedStream）是SparkStreaming的核心抽象，它代表了一個(gè)連續(xù)的數(shù)據(jù)流，可以被看作是RDD的時(shí)間序列。SparkStreaming通過(guò)將流分解為一系列微批處理（micro-batches），并在每個(gè)微批處理上執(zhí)行RDD操作來(lái)處理實(shí)時(shí)數(shù)據(jù)。7.C解析：`join()`是DataFrameAPI中用于根據(jù)指定的鍵將兩個(gè)DataFrame連接（合并）在一起的操作，類似于SQL中的JOIN操作。8.A解析：`repartition(numPartitions)`會(huì)重新計(jì)算數(shù)據(jù)的分布，創(chuàng)建新的分區(qū)，可能會(huì)涉及數(shù)據(jù)的跨節(jié)點(diǎn)復(fù)制。`coalesce(numPartitions)`是`repartition`的一個(gè)特例，當(dāng)減少分區(qū)數(shù)時(shí)，它會(huì)盡量復(fù)用現(xiàn)有的分區(qū)，避免不必要的數(shù)據(jù)移動(dòng)，因此通常比`repartition`更節(jié)省資源。9.B解析：`spark.default.parallelism`是Spark中用于設(shè)置RDD默認(rèn)分區(qū)數(shù)（在沒(méi)有明確指定分區(qū)數(shù)時(shí)）的重要配置參數(shù)，它影響了shuffle操作后的默認(rèn)分區(qū)數(shù)以及某些操作（如`repartition`未指定分區(qū)數(shù)時(shí)）的默認(rèn)分區(qū)數(shù)。10.B解析：`sc.textFile("hdfs://path")`是SparkContext提供的方法，用于從指定的HDFS路徑讀取文本文件，并創(chuàng)建一個(gè)RDD。這是創(chuàng)建RDD的常用方式之一。二、填空題1.parallelize,textFile解析：`parallelize()`方法可以將一個(gè)Scala、Java或Python集合轉(zhuǎn)換為RDD；`textFile()`方法可以從HDFS、HDFS兼容文件系統(tǒng)或本地文件系統(tǒng)中的文本文件創(chuàng)建一個(gè)RDD。2.agg/aggregate解析：在SparkSQL的DataFrameAPI中，`groupBy()`操作后通常需要使用`agg()`或`aggregate()`函數(shù)來(lái)定義聚合操作，例如`count()`,`sum()`,`avg()`等。3.lineage(orDependency)解析：Spark通過(guò)記錄每個(gè)RDD與其父RDD之間的依賴關(guān)系（lineage）來(lái)實(shí)現(xiàn)容錯(cuò)。當(dāng)某個(gè)RDD分區(qū)丟失時(shí)，Spark可以根據(jù)其依賴關(guān)系重新計(jì)算丟失的分區(qū)。4.SparkSession解析：`spark.conf.get("key")`是SparkSession對(duì)象提供的方法，用于獲取當(dāng)前SparkSession的配置信息。5.DStream解析：DStream（DiscretizedStream）是SparkStreaming中代表連續(xù)數(shù)據(jù)流的抽象，可以看作是RDD的時(shí)間序列。6.orderBy解析：`orderBy()`函數(shù)用于對(duì)DataFrame進(jìn)行排序，可以指定排序的列名和排序方式（`asc`或`desc`）。7.Shuffle解析：Shuffle操作是Spark作業(yè)中代價(jià)最高的操作之一，涉及大量數(shù)據(jù)的跨節(jié)點(diǎn)重新分區(qū)和排序。避免不必要的Shuffle可以顯著提升Spark作業(yè)的性能。8.Staging,Scheduling,Execution解析：Spark作業(yè)的執(zhí)行過(guò)程通常包括三個(gè)主要階段：將用戶代碼轉(zhuǎn)換成物理執(zhí)行計(jì)劃（Staging）、將執(zhí)行計(jì)劃提交給調(diào)度器（Scheduling）以及調(diào)度器執(zhí)行物理計(jì)劃（Execution）。9.Text解析：`df.write.saveAsTextFile("hdfs://path")`是DataFrameWriter的`saveAsTextFile()`方法，它將DataFrame按行序列化為文本格式并保存到指定的HDFS路徑。10.Lineage(orDependency)解析：RDD的lineage（依賴關(guān)系）機(jī)制是Spark容錯(cuò)的核心。當(dāng)父RDD丟失且未被緩存時(shí)，Spark會(huì)根據(jù)RDD的依賴樹遞歸地重新計(jì)算丟失的RDD數(shù)據(jù)。三、簡(jiǎn)答題1.簡(jiǎn)述Spark的惰性求值機(jī)制及其優(yōu)點(diǎn)。解析思路：首先解釋什么是惰性求值，即Spark不會(huì)立即執(zhí)行代碼，而是在需要數(shù)據(jù)時(shí)（觸發(fā)行動(dòng)操作）才進(jìn)行計(jì)算。然后說(shuō)明優(yōu)點(diǎn)：1）優(yōu)化執(zhí)行計(jì)劃，Spark可以在觸發(fā)之前分析操作，進(jìn)行優(yōu)化（如謂詞下推、投影下推）；2）減少不必要的計(jì)算，只有被觸發(fā)行動(dòng)操作依賴的數(shù)據(jù)才會(huì)被計(jì)算；3）提高性能，避免了重復(fù)計(jì)算。2.解釋Spark中“數(shù)據(jù)傾斜”的概念，并簡(jiǎn)述一種常見的解決方法。解析思路：首先定義數(shù)據(jù)傾斜，即在進(jìn)行shuffle操作（如`groupByKey()`,`reduceByKey()`,`join()`）時(shí)，某個(gè)鍵（key）對(duì)應(yīng)的數(shù)據(jù)量遠(yuǎn)大于其他鍵，導(dǎo)致該鍵的shuffle任務(wù)耗時(shí)異常長(zhǎng)，成為整個(gè)作業(yè)的瓶頸。然后介紹一種解決方法，例如：對(duì)于`reduceByKey`或聚合操作，可以使用`partitionBy`參數(shù)顯式設(shè)置分區(qū)數(shù)，并使用自定義的分區(qū)函數(shù)，確保數(shù)據(jù)更均勻地分布到各個(gè)分區(qū)；對(duì)于`groupByKey`，可以改用`reduceByKey`或`aggregateByKey`；對(duì)于join操作，可以嘗試broadcast小表，或者使用map-sidejoin。3.比較SparkRDDAPI和DataFrame/DatasetAPI的優(yōu)缺點(diǎn)。解析思路：優(yōu)點(diǎn)（DataFrame/Dataset）：*更豐富的數(shù)據(jù)操作能力：支持SQL語(yǔ)法、復(fù)雜的數(shù)據(jù)類型（如struct,array,map）、窗口函數(shù)等。*更好的優(yōu)化：SparkCatalyst優(yōu)化器可以對(duì)DataFrame/Dataset的操作計(jì)劃進(jìn)行深度優(yōu)化（如謂詞下推、投影下推、列剪裁等），性能通常優(yōu)于RDDAPI。*易于使用和理解：API更接近SQL或JDBC，對(duì)有SQL基礎(chǔ)的開發(fā)者更友好。*強(qiáng)類型檢查：Dataset提供了強(qiáng)類型檢查，減少運(yùn)行時(shí)錯(cuò)誤。缺點(diǎn)（DataFrame/Dataset）：*透明度較低：開發(fā)者無(wú)法像RDD那樣精細(xì)地控制底層的物理執(zhí)行計(jì)劃。*對(duì)某些操作的限制：某些底層的RDD操作在DataFrame/DatasetAPI中可能沒(méi)有直接對(duì)應(yīng)。優(yōu)點(diǎn)（RDDAPI）：*透明度高：可以精確控制數(shù)據(jù)流和執(zhí)行細(xì)節(jié)。*功能全面：支持所有底層的RDD操作。*適用于特定場(chǎng)景：某些需要精細(xì)控制或使用特定RDD原語(yǔ)（如coalesce）的場(chǎng)景。缺點(diǎn)（RDDAPI）：*性能優(yōu)化有限：缺乏Catalyst優(yōu)化器的支持，性能優(yōu)化主要靠手動(dòng)調(diào)優(yōu)。*易出錯(cuò)：缺乏強(qiáng)類型檢查，API相對(duì)復(fù)雜。*不支持SQL：不能直接使用SQL語(yǔ)法。四、編程題1.（15分）編寫PySpark代碼，實(shí)現(xiàn)以下功能：```pythonfrompysparkimportSparkContextsc=SparkContext(appName="LogWordCount")input_path="/user/data/input/log.txt"output_path="/user/data/output"#1.讀取文本文件，創(chuàng)建RDDlog_rdd=sc.textFile(input_path)#2.分詞，創(chuàng)建新RDDwords_rdd=log_rdd.flatMap(lambdaline:line.split(""))#3.統(tǒng)計(jì)詞頻，按詞頻降序排序，取前10word_counts=words_rdd.map(lambdaword:(word,1))\.reduceByKey(lambdaa,b:a+b)\.sortBy(lambdax:x[1],ascending=False)\.take(10)#4.輸出到HDFS（注意：take()返回列表，不能直接寫入HDFS一行）#需要將結(jié)果轉(zhuǎn)換為字符串格式再寫入importjsonoutput_rdd=sc.parallelize([json.dumpsWC(wc)forwcinword_counts])output_rdd.saveAsTextFile(output_path)defjson.dumpsWC(wc):returnf"{wc[0]}:{wc[1]}"```*注意：PySpark本身沒(méi)有內(nèi)置的`json.dumps`直接用于元組，這里定義了一個(gè)簡(jiǎn)單的`json.dumpsWC`函數(shù)將元組轉(zhuǎn)換為字符串。在實(shí)際應(yīng)用中可能需要更復(fù)雜的序列化邏輯。*2.（30分）編寫PySpark代碼，完成DataFrame任務(wù)：```pythonfrompyspark.sqlimportSparkSessionfrompyspark.sql.functionsimportavg,colspark=SparkSession(appName="DataFrameExample")#假設(shè)df已經(jīng)存在，包含user_id,product_id,score列#示例：創(chuàng)建一個(gè)DataFramedata=[(1,101,4.5),(1,102,3.5),(2,101,5.0),(2,103,4.0),(3,102,4.0