2025年(數(shù)據科學與大數(shù)據技術)分析應用試題及答案一、單項選擇題（每題2分，共20分。每題只有一個正確答案，請將正確選項字母填在括號內）1.在Hadoop生態(tài)中，負責集群資源管理與任務調度的組件是（）A.HDFS??B.YARN??C.Hive??D.Pig答案：B2.下列關于數(shù)據湖與數(shù)據倉庫的描述，正確的是（）A.數(shù)據湖僅支持結構化數(shù)據B.數(shù)據倉庫schemaonreadC.數(shù)據湖通常采用schemaonwriteD.數(shù)據湖可原生保留原始格式數(shù)據答案：D3.在SparkRDD轉換操作中，不會觸發(fā)作業(yè)提交的是（）A.map??B.reduceByKey??C.count??D.saveAsTextFile答案：A4.使用Kmeans聚類時，若初始中心點選擇不當導致空簇，最佳應對策略是（）A.降低K值??B.重新運行一次即可C.采用Kmeans++初始化??D.改用DBSCAN答案：C5.在Flink的時間語義中，用于處理亂序事件且可觸發(fā)窗口計算的是（）A.ProcessingTime??B.IngestionTimeC.EventTime??D.SystemTime答案：C6.下列Python庫中，用于實現(xiàn)梯度提升樹且支持類別特征直接輸入的是（）A.XGBoost??B.LightGBM??C.CatBoost??D.scikitlearn答案：C7.在Hive中，將ORC表轉換為分區(qū)表并同時修復元數(shù)據的命令是（）A.ALTERTABLE…ADDPARTITIONB.MSCKREPAIRTABLEC.LOADDATAINPATHD.INSERTOVERWRITE答案：B8.當使用PCA降維時，主成分方差貢獻率之和應（）A.≤1??B.≥1??C.=1??D.與維度無關答案：A9.在Kafka中，保證同一分區(qū)消息順序性的機制是（）A.ConsumerGroup??B.PartitionKeyC.OffsetCommit??D.ReplicationFactor答案：B10.若某電商用戶行為日志文件大小為2.7TB，采用128MB的HDFS塊大小，則理論存儲塊數(shù)量為（）A.17280??B.21600??C.27000??D.13500答案：B解析：2.7TB=2.710241024MB，除以128MB得21600。二、多項選擇題（每題3分，共15分。每題至少有兩個正確答案，多選、少選、錯選均不得分）11.下列屬于NoSQL數(shù)據庫CAP權衡中“放棄強一致性”的典型系統(tǒng)有（）A.MongoDB??B.Cassandra??C.HBase??D.DynamoDB答案：BD12.在SparkSQL中，以下函數(shù)可用于處理缺失值的有（）A.fillna??B.dropna??C.na.replace??D.coalesce答案：ABC13.關于A/B測試的假設檢驗，下列說法正確的有（）A.第一類錯誤指拒絕真實原假設B.功效=1–第二類錯誤概率C.P值越小說明差異越顯著D.樣本量越大，檢出微小差異的能力越弱答案：ABC14.在構建實時數(shù)倉時，Lambda架構包含的層次有（）A.BatchLayer??B.SpeedLayer??C.ServingLayer??D.AlphaLayer答案：ABC15.以下屬于圖算法中心性指標的有（）A.Betweenness??B.Closeness??C.Eigenvector??D.Jaccard答案：ABC三、填空題（每空2分，共20分）16.在Spark中，RDD的__________操作用于將元素扁平化，常與map聯(lián)合使用形成flatMap。答案：轉換17.若某決策樹采用基尼系數(shù)作為劃分標準，則節(jié)點基尼系數(shù)越小表示__________越純。答案：節(jié)點18.在FlinkCEP庫中，模式序列“a+”表示事件a至少連續(xù)出現(xiàn)__________次。答案：119.Hive默認的底層執(zhí)行引擎是__________，從2.x起可切換為Spark或Tez。答案：MapReduce20.在Kafka0.11之后，實現(xiàn)冪等生產者的關鍵參數(shù)是__________。答案：enable.idempotence21.使用pandas讀取CSV時，若文件編碼為GBK，需設置參數(shù)encoding=__________。答案：gbk22.在SQL中，窗口函數(shù)row_number()over(partitionbyuser_idorderbyts)的作用是__________。答案：為每個用戶按時間戳排序生成連續(xù)行號23.若某深度學習模型訓練集準確率達99.9%，驗證集僅85%，最可能發(fā)生了__________。答案：過擬合24.在HDFS中，NameNode的元數(shù)據鏡像文件稱為__________。答案：fsimage25.采用BoxCox變換前，需保證輸入變量取值嚴格__________。答案：大于零四、判斷題（每題1分，共10分。正確打“√”，錯誤打“×”）26.SparkStreaming的批間隔越短，延遲越低，但吞吐量一定提升。答案：×27.在梯度下降中，使用動量項可加速收斂并抑制震蕩。答案：√28.Hive支持事務的前提條件是表必須為ORC格式并開啟事務屬性。答案：√29.對于高基數(shù)類別變量，OneHot編碼總是優(yōu)于TargetEncoding。答案：×30.在Flink中，Checkpoint與Savepoint的實現(xiàn)機制完全相同，可任意互換。答案：×31.數(shù)據傾斜發(fā)生時，增大Reduce數(shù)量一定能緩解熱點。答案：×32.XGBoost的目標函數(shù)由損失函數(shù)與正則項兩部分組成。答案：√33.在Kafka中，Consumer的Offset由Broker強制維護，無法自定義存儲。答案：×34.使用ElasticSearch聚合查詢時，桶聚合與指標聚合可嵌套使用。答案：√35.在Python多進程編程中，由于GIL存在，多進程無法利用多核CPU。答案：×五、簡答題（共25分）36.（封閉型，6分）簡述HDFS寫入流程中數(shù)據節(jié)點故障時的恢復機制。答案：客戶端寫入時，若某DataNode失效，Pipeline會立即關閉，NameNode會接收失敗報告并重新選擇新的DataNode替代，客戶端從未完成處繼續(xù)寫入，剩余包重新構建Pipeline，整個過程對上層透明，保證數(shù)據一致性。37.（開放型，9分）某短視頻平臺日活1.2億，擬構建實時推薦系統(tǒng)。請給出技術選型與理由，涵蓋數(shù)據收集、特征計算、模型推理三層。答案：數(shù)據收集：采用Kafka集群，按用戶分區(qū)，保證順序與吞吐；埋點日志通過Flume或Filebeat實時打入Kafka。特征計算：使用FlinkSQL與CEP完成實時特征，如最近5分鐘觀看類目、滑動速度、點贊率；同時通過Flink的異步I/O關聯(lián)Redis中的用戶畫像。模型推理：選用TensorFlowServing托管深度召回模型與排序模型，特征通過Flink預處理完成后寫入Kafka，再由推薦服務消費并調用TFServing；為降低延遲，采用gRPC接口并開啟批推理，P99延遲控制在30ms內。理由：Kafka提供高吞吐緩沖，F(xiàn)link支持事件時間語義與狀態(tài)管理，TFServing具備熱更新與A/B能力，整體滿足高并發(fā)、低延遲、可擴展需求。38.（封閉型，5分）寫出Spark中廣播變量使用的三步關鍵代碼，并說明其性能優(yōu)勢。答案：valbc=sc.broadcast(smallTable)rdd.map(x=>x.join(bc.value))bc.destroy()優(yōu)勢：避免重復序列化與網絡傳輸，減少Shuffle數(shù)據量，降低GC壓力。39.（開放型，5分）解釋“數(shù)據漂移”與“概念漂移”的區(qū)別，并給出監(jiān)控手段。答案：數(shù)據漂移指輸入特征分布隨時間變化，如用戶年齡分布從年輕群體轉向中年；概念漂移指標簽與特征關系變化，如中年群體購買偏好突變。監(jiān)控：對特征分布采用KolmogorovSmirnov檢驗，設置閾值自動告警；對概念漂移采用滑動窗口比較模型預測概率分布的KL散度，或在線學習算法持續(xù)更新權重。六、計算與分析題（共40分）40.（計算題，10分）某電商大促期間，訂單表order包含字段(order_id,user_id,amount,ts)?，F(xiàn)需統(tǒng)計每10秒窗口的GMV，使用FlinkSQL語法寫出完整語句，并計算當并行度為8、Kafka分區(qū)為16時，每個Subtask平均處理的分區(qū)數(shù)。答案：```sqlCREATETABLEorder_kafka(order_idSTRING,user_idBIGINT,amountDECIMAL(10,2),tsBIGINT,event_timeASTO_TIMESTAMP_LTZ(ts,0),WATERMARKFORevent_timeASevent_timeINTERVAL'2'SECOND)WITH('connector'='kafka','topic'='order','properties.bootstrap.servers'='kafka:9092','scan.startup.mode'='latestoffset','format'='json');CREATEVIEWwindowed_gmvASSELECTTUMBLE_START(event_time,INTERVAL'10'SECOND)ASwindow_start,SUM(amount)ASgmvFROMorder_kafkaGROUPBYTUMBLE(event_time,INTERVAL'10'SECOND);```分區(qū)數(shù)16/并行度8=2個分區(qū)/Subtask。41.（分析題，15分）給定用戶行為日志：user_id,item_id,category,behavior,ts。其中behavior取值pv、buy、cart、fav。要求：1.使用SparkDataFrameAPI計算每個用戶近7天購買轉化率（buy/pv）。2.過濾轉化率>0.05且pv≥100的用戶，輸出其Top20并解釋代碼。答案：```pythonfrompyspark.sqlimportfunctionsasFfrompyspark.sql.windowimportWindowdf=spark.read.parquet("user_log")w=Window.partitionBy("user_id").orderBy(F.col("ts").desc()).rangeBetween(786400,0)pv_cnt=F.sum(F.when(F.col("behavior")=="pv",1).otherwise(0)).over(w)buy_cnt=F.sum(F.when(F.col("behavior")=="buy",1).otherwise(0)).over(w)rate_df=df.withColumn("pv",pv_cnt)\.withColumn("buy",buy_cnt)\.withColumn("ctr",F.col("buy")/F.col("pv"))\.filter((F.col("ctr")>0.05)&(F.col("pv")>=100))\.select("user_id","pv","buy","ctr").distinct()\.orderBy(F.col("ctr").desc()).limit(20)rate_df.write.csv("high_value_user")```解釋：采用窗口函數(shù)避免多次Join，rangeBetween以秒為單位精確近7天；過濾后按轉化率降序取Top20，可直接用于精準營銷。42.（綜合題，15分）某市政府開放數(shù)據包含出租車GPS點，字段：vehicle_id,longitude,latitude,speed,timestamp,occupancy。數(shù)據量3TB，存儲于HDFS，塊大小256MB。任務：a)設計一種基于Spark的分布式算法，檢測任意車輛連續(xù)30秒速度為0且載客狀態(tài)為1的“疑似拒載”事件，輸出車輛ID、開始時間、結束時間、持續(xù)秒數(shù)。b)評估該算法在200臺Executor、每臺4核、16GB內存下的最大并行度與內存占用。答案：a)算法步驟：1.按vehicle_id分區(qū)并按timestamp排序，使用mapPartitions構造滑動窗口；2.維護長度為30秒的隊列，若隊列內所有speed=0且occupancy=1，則觸發(fā)事件；3.輸出事件三元組，reduce合并相鄰窗口避免重復。核心代碼：```scalacaseclassGps(vehicle_id:String,ts:Long,speed:Double,occupancy:Int)valrdd=spark.read.parquet("gps").as[Gps].rdd.groupBy(_.vehicle_id).mapValues(iter=>{valsorted=iter.toArray.sortBy(_.ts)valbuf=scala.collection.mutable.Queue[Gps]()varout=List[(String,Long,Long,Int)]()for(p<sorted){buf.enqueue(p)while(buf.head.ts<p.ts30)buf.dequeue()if(buf.length>1&&buf.forall(_.speed==0)&&buf.head.occupancy==1){out=(p.vehicle_id,buf.head.ts,p.ts,30)::out}}out}).flatMap(_._2).toDF("vehicle_id","start","end","duration").write.parquet("reject_events")```b)資源評估：總塊數(shù)=3TB/256MB≈12288塊；并行度=200Executor4核=800任務并發(fā)；每任務平均處理12288/800≈15塊，約3.75GB數(shù)據；內存占用：每任務序列化對象約保留1.2倍原始數(shù)據，即4.5GB，小于Executor可用內存16GB0.6≈9.6GB，滿足要求；若開啟Tungsten優(yōu)化，內存可降至2.5GB，安全余量充足。七、設計題（共20分）43.某頭部快消企業(yè)計劃搭建全域數(shù)據中臺，需整合線下POS、電商OMS、社媒聆聽、ERP四大源，日均增量數(shù)據約5TB，峰值QPS8萬，要求支持實時與離線分析，且合規(guī)審計保留5年。請給出：1.整體架構圖（文字描述即可）；2.數(shù)據分層（貼源、明細、匯總、應用）的存儲格式與壓縮策略；3.實時鏈路ExactlyOnce實現(xiàn)細節(jié)；4.成本估算與降本方案。答案：1.架構：數(shù)據源→Kafka→FlinkCDC→DeltaLake（Bronze）→SparkETL→Iceberg（Silver）→ClickHouse（Gold）→Superset/Tableau；離線補充：Sqoop抽取ERP→HDFS→Hive外部表；元數(shù)據：Atlas統(tǒng)一血緣；安全：Ranger+Kerberos+列加密；備份：冷數(shù)據轉儲至OSS深度歸檔。2.分層：貼源層：DeltaLake原始JSON，ZSTD壓縮，保留3天；明細層：IcebergParquet，SNAPPY，按dt+hour分區(qū)，生命周期90天；匯總層：ClickHouseMergeTree，LZ4，預聚合1分鐘粒度，存180天；應用層：ClickHouse物化視圖，面向指標，SSD存儲，保留1年；冷數(shù)據：ORC+ZSTD存OSS，單位GB月成本0.03元。3.ExactlyOnce：Kafka→Flink采用兩階段提交，Checkpoint與Kafka事務聯(lián)動；Flink→DeltaLake通過checkpoint+idempotentwrite，依賴Delta的樂觀并發(fā)控制；下游ClickHouse采用ReplacingMergeTree去重鍵(order_id,version)。4.成本：峰值8萬QPS，Kafka需30節(jié)點（32核128GB，SSD3.2TB），年費用約180萬；Flink400核，年120萬；存儲5年總量=5TB3655≈6PB，熱數(shù)據10%存SSD單價0.8元/GB/月，冷數(shù)據90%轉OSS0.03元/GB/月，年存儲費≈6PB0.10.812+6PB0.90.0312≈5.8百萬；降本：開啟ZSTD壓縮率提升30%；按業(yè)務降采樣，對社媒文本僅保留關鍵詞向量，減少40%體積；采用彈性Flink集群，夜間縮容50%，節(jié)省25%計算費；總計年成本可降至約4.2百萬。八、編程實現(xiàn)題（共30分）44.使用Python實現(xiàn)分布式訓練腳本，基于Horovod+PyTorch，完成CTR預估任務。數(shù)據集為Criteo1TB子集，包含26個類別特征與13個數(shù)值特征。要求：1.代碼支持多機多卡，參數(shù)服務器架構；2.采用DeepFM模型，類別特征使用Embedding，數(shù)值特征歸一化；3.輸出訓練日志含loss、AUC、學習率，支持TensorBoard；4.給出在2節(jié)點各8卡V100下的訓練耗時與AUC結果。答案：```pythonimporthorovod.torchashvdimporttorch,torch.nnasnnfromtorch.utils.dataimportDataLoaderfromsklearn.preprocessingimportStandardScalerimportnumpyasnphvd.init()torch.cuda.set_device(hvd.local_rank())classDeepFM(nn.Module):def__init__(self,feat_sizes,emb_dim=16,mlp_dims=[256,128,64]):super().__init__()self.emb=nn.ModuleList([nn.Embedding(feat_sizes[i],emb_dim)foriinrange(26)])self.fm_first=nn.ModuleList([nn.Embedding(feat_sizes[i],1)foriinrange(26)])self.mlp=nn.Sequential(nn.Linear(13+26emb_dim,mlp_dims[0]),nn.ReLU(),nn.Linear(mlp_dims[0],mlp_dims[1]),nn.ReLU(),nn.Linear(mlp_dims[1],mlp_dims[2]),nn.ReLU(),nn.Linear(mlp_dims[2],1))defforward(self,dense,sparse):embs=[self.emb[i](sparse[:,i])foriinrange(26)]fm_first=[self.fm_first[i](sparse[:,i])foriinrange(26)]fm_second=sum(embs).pow(2).sum(1,keepdim=True)sum(e.pow(2).sum(1,keepdim=True)foreinembs)dnn_in=torch.cat([dense]+embs,1)dnn_out=self.mlp(dnn_in)logit=dnn_out+0.5fm_second+sum(fm_first)returntorch.sigmoid(logit)dataset=CriteoDataset("train.txt",scaler=StandardScaler())sampler=torch.utils.data.distributed.DistributedSampler(dataset,num_replicas=hvd.size(),rank=hvd.rank())loader=DataLoader(dataset,batch_size=4096,sampler=sampler,num_workers=4)model=DeepFM(feat_sizes).cuda()optimizer=torch.optim.Adam(model.parameters(),lr=0.002hvd.size())optimizer=hvd.DistributedOptimizer(optimizer,named_parameters=d_parameters())hvd.broadcast_parameters(model.state_dict(),root_rank=0)logger=torch.utils.tensorboard.SummaryWriter(log_dir="./logs")ifhvd.rank()==0elseNoneforepochinrange(3):sampler.set_epoch(epoch)forstep,(