2025年P(guān)ython大數(shù)據(jù)技術(shù)考試試卷Hadoop與Spark深度剖析考試時(shí)間：______分鐘總分：______分姓名：______一、簡(jiǎn)答題（每題8分，共40分）1.請(qǐng)簡(jiǎn)述HDFSNameNode的主要職責(zé)及其可能存在的單點(diǎn)故障問題，并說明Hadoop2.x中采用何種機(jī)制緩解該問題。2.描述MapReduce計(jì)算模型中“Shuffle”階段的主要過程及其對(duì)任務(wù)性能的影響，并解釋Combiner階段的作用。3.解釋Spark中RDD的“惰性計(jì)算”特性及其帶來的優(yōu)勢(shì)，并說明RDD的“l(fā)ineage”（血統(tǒng)）概念在容錯(cuò)機(jī)制中的作用。4.比較SparkDataFrame和Dataset的主要區(qū)別，并說明為什么Dataset通常比DataFrame提供更好的性能和安全性。5.簡(jiǎn)述SparkStreaming處理實(shí)時(shí)數(shù)據(jù)流的基本原理，并說明其如何將連續(xù)的數(shù)據(jù)流轉(zhuǎn)換為離散的數(shù)據(jù)RDD。二、綜合應(yīng)用題（第1題15分，第2題25分，共40分）1.假設(shè)你已經(jīng)通過PySpark啟動(dòng)了一個(gè)SparkSession，名為`spark`?，F(xiàn)有一個(gè)存儲(chǔ)在HDFS上的CSV文件`/user/data/sales.csv`，該文件包含字段`date`（日期字符串）、`product`（產(chǎn)品名稱）和`amount`（銷售額整數(shù)）。請(qǐng)編寫PySpark代碼完成以下任務(wù)：*讀取該CSV文件，創(chuàng)建一個(gè)DataFrame，字段名分別為`Date`,`Product`,`Amount`。*篩選出`Amount`大于1000的記錄。*按照產(chǎn)品名稱`Product`進(jìn)行分組，并對(duì)每個(gè)產(chǎn)品的總銷售額`Amount`進(jìn)行排序。*將結(jié)果保存為名為`sales_summary`的Parquet文件到`/user/output/`目錄下。2.假設(shè)你需要使用SparkStreaming處理來自Kafka的主題為`sensor_data`的實(shí)時(shí)傳感器數(shù)據(jù)流，數(shù)據(jù)格式為每行一個(gè)JSON字符串，包含字段`timestamp`（時(shí)間戳字符串）和`temperature`（溫度浮點(diǎn)數(shù)）。請(qǐng)編寫PySpark代碼完成以下任務(wù)：*配置SparkStreaming連接到Kafka，讀取`sensor_data`主題的數(shù)據(jù)。*將接收到的JSON字符串解析為包含`timestamp`和`temperature`字段的DataFrame。*計(jì)算過去5分鐘內(nèi)所有溫度數(shù)據(jù)的平均值，并將結(jié)果實(shí)時(shí)輸出到控制臺(tái)（每10秒輸出一次最新計(jì)算結(jié)果）。*要求使用DataFrameAPI進(jìn)行數(shù)據(jù)處理。試卷答案一、簡(jiǎn)答題1.HDFSNameNode的主要職責(zé)包括：管理文件系統(tǒng)的元數(shù)據(jù)（目錄結(jié)構(gòu)、文件塊位置信息等），維護(hù)命名空間，處理客戶端對(duì)文件的訪問請(qǐng)求（如打開、讀取、創(chuàng)建、刪除文件等），以及協(xié)調(diào)DataNode之間的數(shù)據(jù)塊復(fù)制。單點(diǎn)故障問題：NameNode是HDFS集群的管理核心，其故障會(huì)導(dǎo)致整個(gè)HDFS服務(wù)不可用，影響上層應(yīng)用。緩解機(jī)制：Hadoop2.x引入了HighAvailability（HA）機(jī)制，通常采用兩個(gè)NameNode（Active和Standby）配合一個(gè)共享的元數(shù)據(jù)存儲(chǔ)（JournalNode或QuorumJournalManager），通過心跳機(jī)制實(shí)現(xiàn)故障自動(dòng)切換。2.Shuffle階段主要過程：在MapTask完成后，將Map階段的輸出（中間KV對(duì)）根據(jù)ReduceTask的輸入鍵進(jìn)行排序，然后將排序好的數(shù)據(jù)塊（Partition）傳輸?shù)綄?duì)應(yīng)的ReduceTask所在的TaskTracker/Executor上。對(duì)性能影響：Shuffle是MapReduce作業(yè)中CPU和I/O消耗最大的階段之一，其效率和速度直接影響整個(gè)作業(yè)的執(zhí)行時(shí)間。數(shù)據(jù)傳輸量越大、網(wǎng)絡(luò)帶寬越低、排序和寫入磁盤開銷越大，性能瓶頸越明顯。Combiner階段作用：Combiner是一個(gè)局部Reduce操作，它在MapTask輸出數(shù)據(jù)到ReduceTask之前，在每個(gè)MapTask所在的節(jié)點(diǎn)上執(zhí)行一次本地Reduce，旨在減少網(wǎng)絡(luò)傳輸?shù)臄?shù)據(jù)量，優(yōu)化Shuffle階段的開銷。它通常是一個(gè)輕量級(jí)的Reduce，計(jì)算代價(jià)小。3.RDD的“惰性計(jì)算”特性：Spark對(duì)用戶編寫的RDD轉(zhuǎn)換操作（如`map`,`filter`等）并不會(huì)立即執(zhí)行，而是將這些操作組織成一個(gè)DirectAcyclicGraph（DAG），只在執(zhí)行動(dòng)作操作（如`collect`,`count`,`save`等）時(shí)才觸發(fā)計(jì)算，并執(zhí)行DAG中的必要轉(zhuǎn)換。優(yōu)勢(shì)：減少了不必要的計(jì)算，提高了性能；通過DAG優(yōu)化，Spark可以應(yīng)用如Tungsten執(zhí)行引擎、內(nèi)存管理等優(yōu)化策略；方便調(diào)試，因?yàn)榭梢杂^察到實(shí)際執(zhí)行的計(jì)算圖。RDD的“l(fā)ineage”在容錯(cuò)中的作用：RDD的lineage是指一個(gè)RDD通過一系列轉(zhuǎn)換操作從其父RDD派生出來的過程記錄。當(dāng)某個(gè)RDD分區(qū)在計(jì)算過程中丟失時(shí)，Spark可以通過其lineage信息，反向追蹤到其父RDD，并重新計(jì)算丟失的分區(qū)，從而實(shí)現(xiàn)容錯(cuò)恢復(fù)。4.SparkDataFrame與Dataset的主要區(qū)別：類型系統(tǒng)：DataFrame是分布式數(shù)據(jù)集合，其每一列有一個(gè)明確的名稱和數(shù)據(jù)類型，本質(zhì)上是Row對(duì)象的集合。Dataset是DataFrame的泛化，它不僅包含列名和數(shù)據(jù)類型，還包含類型信息（TypeInformation），是一個(gè)特定Java/Scala/Python類型（CaseClass/Tuple/PythonClass）的分布式集合。性能：Dataset在運(yùn)行時(shí)利用了類型擦除和編譯器優(yōu)化（如內(nèi)聯(lián)函數(shù)調(diào)用），通常比DataFrame有更好的執(zhí)行效率，尤其是在涉及復(fù)雜表達(dá)式和優(yōu)化時(shí)。Dataset還可以利用Java/Scala的強(qiáng)類型檢查能力提前發(fā)現(xiàn)錯(cuò)誤。API：DataFrame提供統(tǒng)一的DataFrameAPI，跨語言支持較好。Dataset提供了類型安全的API，但相對(duì)DataFrameAPI更少，跨語言支持也較弱（如PySpark中只有DataFrameAPI，沒有原生DatasetAPI）。安全性：Dataset通過類型信息可以在運(yùn)行時(shí)檢查API調(diào)用的正確性，提供了更強(qiáng)的API使用安全性。5.SparkStreaming處理實(shí)時(shí)數(shù)據(jù)流的基本原理：SparkStreaming將實(shí)時(shí)數(shù)據(jù)流視為一系列離散的數(shù)據(jù)小批量（micro-batches），并將每個(gè)小批量作為一個(gè)RDD進(jìn)行處理。它通過持續(xù)的從數(shù)據(jù)源（如Kafka,Flume,Socket等）讀取數(shù)據(jù)，將數(shù)據(jù)分批追加到一個(gè)新的RDD中，然后利用Spark的核心RDD計(jì)算能力，對(duì)每個(gè)小批量應(yīng)用相同的Spark作業(yè)（如SparkSQL查詢、DataFrame轉(zhuǎn)換等），從而實(shí)現(xiàn)近乎實(shí)時(shí)的數(shù)據(jù)處理。將連續(xù)數(shù)據(jù)流轉(zhuǎn)換為離散RDD的方式：SparkStreaming使用一個(gè)名為DStream（DiscretizedStream）的抽象，它代表了連續(xù)數(shù)據(jù)流的一個(gè)不可變、時(shí)間有序的RDD序列。每個(gè)DStream操作（如`map`,`reduce`,`window`等）都會(huì)產(chǎn)生一個(gè)新的DStream，該DStream的每個(gè)元素都是一個(gè)包含對(duì)應(yīng)時(shí)間區(qū)間內(nèi)數(shù)據(jù)的RDD。通過這種方式，連續(xù)的數(shù)據(jù)流被分解成一系列可以在Spark中進(jìn)行批處理的小批次RDD。二、綜合應(yīng)用題1.```pythonfrompyspark.sqlimportSparkSessionfrompyspark.sql.functionsimportcolspark=SparkSession.builder.appName("SalesSummary").getOrCreate()#讀取CSV文件，指定分隔符，忽略標(biāo)題行，并設(shè)置字段名df=spark.read.csv("/user/data/sales.csv",sep=",",header=True,inferSchema=True,schema="Datestring,Productstring,Amountinteger")#篩選Amount大于1000的記錄filtered_df=df.filter(col("Amount")>1000)#按Product分組，并對(duì)每個(gè)產(chǎn)品的Amount總和進(jìn)行排序（默認(rèn)升序，使用desc降序）grouped_sorted_df=filtered_df.groupBy("Product").agg({"Amount":"sum"}).orderBy(col("sum(Amount)").desc())#將結(jié)果保存為Parquet文件grouped_sorted_df.write.parquet("/user/output/sales_summary")spark.stop()```2.```pythonfrompyspark.sqlimportSparkSessionfrompyspark.sql.functionsimportfrom_json,col,meanfrompyspark.sql.typesimportStructType,StructField,StringType,FloatTypespark=SparkSession.builder.appName("SensorStreaming").getOrCreate()#定義輸入數(shù)據(jù)的JSON格式結(jié)構(gòu)schema=StructType([StructField("timestamp",StringType(),True),StructField("temperature",FloatType(),True)])#從Kafka讀取數(shù)據(jù)流，指定主題，Kafka服務(wù)器等（此處為示例配置）df=spark.readStream.format("kafka").option("kafka.bootstrap.servers","kafka_server:9092").option("subscribe","sensor_data").load()#解析JSON數(shù)據(jù)json_df=df.selectExpr("CAST(valueASSTRING)").select(from_json(col("value"),schema).alias("data")).select("data.*")#定義一個(gè)窗口，計(jì)算過去5分鐘內(nèi)所有溫度的平均值，窗口間隔10秒滑動(dòng)windowed_df=json_df.withColumn("timestamp",col("timestamp").cast("timestamp")).groupBy(window(col("timestamp"),"5minute