如果你最近在學(xué) Spark又想找一個不繞彎、不燒集群的實戰(zhàn)練習(xí)這一組 Spark 小練習(xí)應(yīng)該正合你意。我最近剛把它完整過了一遍用 PySpark 讀一份本地生成的銷售訂單 CSV接著做數(shù)據(jù)清洗、去重和日期規(guī)整再用 DataFrame API 與 Spark SQL 分別算出“每個月、每個品類的訂單量、銷售額、平均客單價”最后把結(jié)果寫出并回讀校驗。整個流程不依賴分布式集群單機(jī)就能跑也不需要你提前會 Scala。它沒有多高深但對靠刷題入門的人來說能一次性補(bǔ)上最關(guān)鍵的“從文件到 DataFrame再從 DataFrame 到結(jié)果報表”的完整手感。這個小練習(xí)解決的是一個很典型的痛點很多人把 RDD 算子背得滾瓜爛熟可真給一份 CSV卻不知道怎么下手也有人只會寫 Spark SQL但完全不理解 DataFrame 算子與 SQL 的對應(yīng)關(guān)系。做完這組練習(xí)你會對groupBy、agg、withColumn、filter這些概念有具體體感也能看懂 Spark UI 里最簡單的 Job 和 Stage 長什么樣。如果你是剛學(xué)完 Spark 基礎(chǔ)的學(xué)生或者準(zhǔn)備大數(shù)據(jù)開發(fā)面試但缺實操的初學(xué)者可以照著下面的方案完整走一遍如果已經(jīng)能用 Spark 做生產(chǎn)任務(wù)也可以重點看后面的問題排查部分。1. 動手前先拆解這個小練習(xí)到底練什么1.1 練習(xí)目標(biāo)不是搭集群而是跑通一條完整的 Spark SQL 數(shù)據(jù)鏈路我第一次做這個練習(xí)時差點被“分布式”“集群”這些詞帶偏想著要不要先搭一個三節(jié)點環(huán)境。后來我意識到對一個以學(xué)習(xí)為目的的小練習(xí)來說最關(guān)鍵的不是機(jī)器數(shù)量而是你能不能把一條數(shù)據(jù)處理鏈路完整地走通。所謂鏈路在這里就是五個環(huán)節(jié)讀入外部數(shù)據(jù)、觀察數(shù)據(jù)形態(tài)、做清洗轉(zhuǎn)換、執(zhí)行聚合統(tǒng)計、把結(jié)果寫回文件系統(tǒng)。我把每個環(huán)節(jié)對應(yīng)到具體動作第一步用spark.read.csv把本地 CSV 變成DataFrame第二步用printSchema和show看字段類型與樣例數(shù)據(jù)第三步做狀態(tài)過濾、訂單去重和日期字段拆分第四步用groupBy與agg算出每個品類每個月的訂單量和銷售額第五步把結(jié)果寫成一個 CSV 文件再重新讀回來驗證。五個環(huán)節(jié)依次走完Spark 里最常用的二十幾個方法基本都能覆蓋到后續(xù)再學(xué)窗口函數(shù)、Join、Streaming 時也不會覺得基礎(chǔ)是懸空的。1.2 為什么選 PySpark 加 Spark SQL 的方式而不是純 RDD 練習(xí)真的建議新手從 PySpark 的 DataFrame 和 Spark SQL 入手而不是把純 RDD 算子當(dāng)作主練習(xí)。原因很簡單RDD 的很多底層算子實際開發(fā)中用得越來越少而 DataFrame 背后的 Catalyst 優(yōu)化器能自動做謂詞下推、列裁剪等優(yōu)化這才是日常處理結(jié)構(gòu)化數(shù)據(jù)的核心方式。用 PySpark 寫練習(xí)還有一個實際優(yōu)勢你在交互環(huán)境里寫 Scala每按一次回車等著編譯切到 Python 后可以快速df.show()看結(jié)果、改邏輯、再運(yùn)行試錯成本低很多。有些人總覺得“生產(chǎn)環(huán)境用 Scala所以練習(xí)也要 Scala”這個說法至少在這個階段不成立。你用 Scala 還是 Python只要最終生成的是同樣的 DataFrame 和 SQL 任務(wù)物理執(zhí)行計劃不會有本質(zhì)差別。練習(xí)的核心應(yīng)該放在算子的語義和數(shù)據(jù)的流轉(zhuǎn)上而不是被語言語法干擾。等你以后真去改 Spark 源碼或者做底層性能優(yōu)化再學(xué) Scala 也來得及。2. 五分鐘搭好本地環(huán)境并生成一份模擬訂單數(shù)據(jù)2.1 本地環(huán)境搭建JDK、虛擬環(huán)境和 pip 安裝的省心組合本地跑 PySpark 其實比想象中簡單。我先說版本組合我用的 Spark 3.5.x配合 Python 3.9JDK 用 17。你用 JDK 8 或 11 也行但不建議直接用最新的 JDK 21因為部分 Spark 版本還不支持。安裝步驟可以壓縮成三步先確保java -version能輸出版本號再創(chuàng)建一個 Python 虛擬環(huán)境最后在虛擬環(huán)境里用 pip 安裝 PySpark。python -m venv spark_practice source spark_practice/bin/activate # Windows 下是 spark_practice\Scripts\activate pip install pyspark3.5.1 pandas這里我特別想提醒一句一定要固定版本。我見過有人直接pip install pyspark裝到最新版結(jié)果跟著舊教程寫代碼API 對不上最后一堆報錯。指定版本號后至少你能確定當(dāng)前代碼所對應(yīng)的行為。還有一點pip 安裝的 PySpark 會自帶本地運(yùn)行所需的基礎(chǔ)組件你不需要單獨裝一份 Hadoop啟動local[*]模式時會直接以本地多線程方式模擬分布式執(zhí)行。2.2 用腳本生成五萬行訂單 CSV讓練習(xí)有真實感不要隨便在網(wǎng)上找一份不知道格式的數(shù)據(jù)來練最好自己用腳本生成一份可控制的模擬訂單數(shù)據(jù)。字段設(shè)計要有電商業(yè)務(wù)的真實感但又要方便后續(xù)聚合訂單號、用戶ID、下單日期、商品類目、金額、區(qū)域、訂單狀態(tài)。我生成五萬行日期分布在一到四月份金額在 10 到 5000 之間狀態(tài)里有已完成、已取消、已退款。這樣的數(shù)據(jù)既能做過濾也能做按月的分組統(tǒng)計。下面這個腳本可以直接保存成generate_data.py運(yùn)行import random import datetime import csv random.seed(42) categories [手機(jī)數(shù)碼, 家用電器, 服飾鞋包, 食品生鮮, 美妝個護(hù)] regions [華東, 華北, 華南, 西南, 東北] status_list [已完成, 已取消, 已退款] with open(data/sales.csv, w, newline, encodingutf-8) as f: writer csv.writer(f) writer.writerow([order_id, user_id, order_date, category, amount, region, status]) for i in range(50000): days random.randint(0, 120) order_date datetime.date(2024, 1, 1) datetime.timedelta(daysdays) writer.writerow([ fO{i:06d}, fU{random.randint(1, 2000):05d}, order_date.isoformat(), random.choice(categories), round(random.uniform(10, 5000), 2), random.choice(regions), random.choice(status_list) ])寫文件時千萬記得帶換行處理參數(shù)和 UTF-8 編碼。如果你在 Windows 上不小心用了系統(tǒng)默認(rèn)編碼后面 Spark 讀出來中文可能直接亂碼。生成之后可以先看前幾行確認(rèn)是正常 CVS 結(jié)構(gòu)再進(jìn)入下一步。3. 核心實操從讀 CSV 到出報表的完整鏈路3.1 初始化 SparkSession把 CSV 變成 DataFrame在 PySpark 里所有功能的入口都是SparkSession。小練習(xí)不需要花里胡哨的配置我通常只設(shè)置三個東西應(yīng)用名、運(yùn)行模式、shuffle 分區(qū)數(shù)。運(yùn)行模式寫成local[*]表示使用本機(jī)所有可用線程spark.sql.shuffle.partitions設(shè)置為 8是為了避免默認(rèn) 200 個 shuffle 分區(qū)在小數(shù)據(jù)量下造成大量空任務(wù)。from pyspark.sql import SparkSession spark (SparkSession.builder .appName(spark-small-practice) .master(local[*]) .config(spark.sql.shuffle.partitions, 8) .getOrCreate()) df spark.read.csv(data/sales.csv, headerTrue, inferSchemaTrue) df.printSchema() df.show(5, truncateFalse)inferSchemaTrue會讓 Spark 自動推斷每列類型對練習(xí)很方便。不過生產(chǎn)環(huán)境不要依賴這個參數(shù)一是它需要額外掃描一遍數(shù)據(jù)二是推斷結(jié)果可能和你預(yù)期不一致。更好的做法是像下面這樣顯式定義 Schema尤其在實際團(tuán)隊協(xié)作時數(shù)據(jù)類型的穩(wěn)定性比代碼少幾行重要得多。from pyspark.sql.types import (StructType, StructField, StringType, IntegerType, DoubleType, DateType) sales_schema StructType([ StructField(order_id, StringType()), StructField(user_id, StringType()), StructField(order_date, StringType()), StructField(category, StringType()), StructField(amount, DoubleType()), StructField(region, StringType()), StructField(status, StringType()) ])這一步做完df就是一張“可以橫著拆開、豎著切分”的表。建議你花兩分鐘把printSchema的輸出抄在筆記里后面寫代碼時時刻對照字段類型能省掉不少因類型誤用導(dǎo)致的報錯。3.2 數(shù)據(jù)清洗先過濾臟數(shù)據(jù)和重復(fù)訂單很多人一拿到 DataFrame 就急著groupBy這是練習(xí)里最容易踩的坑。真實的訂單數(shù)據(jù)一定有取消單可能還有重復(fù)入庫的臟數(shù)據(jù)如果你不過濾掉聚合結(jié)果就會和業(yè)務(wù)口徑對不上。我在這份模擬數(shù)據(jù)里只統(tǒng)計“已完成”的訂單同時按order_id去重再排除金額為空或金額小于等于 0 的記錄。日期字段讀取后是字符串所以還得先轉(zhuǎn)成真正的日期類型再拆出年份、月份和星期。from pyspark.sql import functions as F df_clean (df .filter(df[status] 已完成) .dropDuplicates([order_id]) .withColumn(order_date, F.to_date(order_date, yyyy-MM-dd)) .withColumn(year, F.year(order_date)) .withColumn(month, F.month(order_date)) .withColumn(week_of_year, F.weekofyear(order_date)) .filter(F.col(amount).isNotNull() (F.col(amount) 0))) print(清洗前記錄數(shù), df.count()) print(清洗后記錄數(shù), df_clean.count())你可以用這個count對比直觀感受一下什么叫“數(shù)據(jù)質(zhì)量影響結(jié)果”。在這個小練習(xí)里過濾前后甚至能差出幾千條訂單如果直接把臟數(shù)據(jù)也算進(jìn)銷售額最后報表一定會被挑戰(zhàn)。關(guān)于withColumn我想多說一句它是生成新列不會修改原來的DataFrame所以鏈?zhǔn)秸{(diào)用時每步都必須接收返回值。我看到有初學(xué)者寫df.withColumn(...)后以為原表變了下一行繼續(xù)用df結(jié)果新列永遠(yuǎn)不生效。這是 PySpark 和 Pandas 在語義上一個很大的區(qū)別。3.3 分組聚合DataFrame API 和 Spark SQL 各寫一遍清洗完成之后就到了練習(xí)的重點計算每個月份、每個品類的訂單量、總銷售額和平均客單價。我建議你用兩種方式各寫一遍先看 DataFrame 寫法再看 Spark SQL 寫法。DataFrame 寫法如下result_df (df_clean .groupBy(category, month) .agg( F.count(order_id).alias(order_cnt), F.round(F.sum(amount), 2).alias(total_amount), F.round(F.avg(amount), 2).alias(avg_amount) ) .orderBy(month, F.desc(total_amount))) result_df.show(30, truncateFalse)這里groupBy里的列對應(yīng) SQL 的GROUP BYagg里的每一行則是一個聚合表達(dá)式。alias用來給輸出列起名別小看這個動作生產(chǎn)報表里下游經(jīng)常靠列名對接名字混亂很麻煩。Spark SQL 寫法則是先把DataFrame注冊成臨時視圖再直接寫標(biāo)準(zhǔn) SQLdf_clean.createOrReplaceTempView(sales) sql_result spark.sql( SELECT category, month, COUNT(order_id) AS order_cnt, ROUND(SUM(amount), 2) AS total_amount, ROUND(AVG(amount), 2) AS avg_amount FROM sales GROUP BY category, month ORDER BY month, total_amount DESC ) sql_result.show(30, truncateFalse)很多同學(xué)會問既然結(jié)果一樣練一種不就好了我的看法是這兩種寫法你要至少各練一遍因為工作里一半的人用 DataFrame API另一半寫 SQL你看不懂任何一方都會很吃力。更關(guān)鍵的是你得在腦子里建立起兩者的對應(yīng)關(guān)系SELECT對應(yīng)selectWHERE對應(yīng)filterGROUP BY對應(yīng)groupByORDER BY對應(yīng)orderBy。建立不了這個映射你讀別人代碼就像看天書。寫完后可以用下面的方式快速驗證兩個結(jié)果是否一致避免自己騙自己diff_count result_df.exceptAll(sql_result).count() print(兩張結(jié)果表的差異行數(shù), diff_count)3.4 寫出結(jié)果并回讀校驗確保不是“看起來對”Spark 寫 CSV 和 Pandas 寫 CSV 有一個很不一樣的地方Spark 寫出的不是單個文件而是一個目錄里面會有多個part-*文件外加一個空白的_SUCCESS標(biāo)記文件。這個設(shè)計是為了并行寫出但在小練習(xí)里我們通常希望最終能看到一個可讀的單文件。所以我會在寫出前用repartition(1)把結(jié)果合并到一個分區(qū)output_path output/sales_summary sql_result.repartition(1).write.mode(overwrite).option(header, True).csv(output_path)這里要說明一下repartition(1)是重新落一次全量數(shù)據(jù)數(shù)據(jù)量大的時候會造成后續(xù)任務(wù)單分區(qū)處理、非常慢甚至 OOM練習(xí)里數(shù)據(jù)量小無所謂千萬別把它當(dāng)成生產(chǎn)環(huán)境寫單文件的通用手段。如果生產(chǎn)上確實需要單文件一般先正常寫出再單獨做文件合并。寫完以后不要直接打開 CSV 看一眼就說完成更嚴(yán)謹(jǐn)?shù)淖龇ㄊ怯?Spark 把輸出再讀回來和內(nèi)存里的結(jié)果做對比check_df spark.read.csv(output_path, headerTrue, inferSchemaTrue) print(輸出文件記錄數(shù), check_df.count()) check_df.show(10, truncateFalse)這樣能發(fā)現(xiàn)很多隱藏問題。比如字段順序變了、數(shù)值被截斷、分區(qū)文件太多都可以在回讀階段暴露。我自己的習(xí)慣是任何一次作業(yè)寫到文件系統(tǒng)之后都要立刻讀一遍哪怕只是打印前二十行也不要跳過去。3.5 用 explain 和 Spark UI 驗證執(zhí)行細(xì)節(jié)跑通結(jié)果只是第一步我并不建議你停在“能出數(shù)”就結(jié)束。如果身邊有人帶你他大概率會問你一句你這個任務(wù)到底分成了幾個 Stage有沒有 Shuffle要回答這個問題最簡單的方式是看執(zhí)行計劃。在 PySpark 里可以對一個 DataFrame 調(diào)用explainsql_result.explain(modeextended)不要被一長串輸出嚇到你只需要關(guān)注幾個點掃描數(shù)據(jù)源有沒有做列裁剪和謂詞下推聚合過程中有沒有出現(xiàn)Exchange節(jié)點也就是 Shuffle最后的結(jié)果有沒有走排序。對這個小練習(xí)來說GROUP BY一定會觸發(fā) Shuffle所以你會看到Exchange這是正常的。如果還想看更直觀的進(jìn)度運(yùn)行作業(yè)時打開瀏覽器訪問http://localhost:4040也就是 Spark UI。你能看到 Job、Stage、Executor 的基本情況。本地練習(xí)雖然遠(yuǎn)沒有集群環(huán)境那么壯觀但盡早培養(yǎng)看 UI 的習(xí)慣對以后排查慢任務(wù)、數(shù)據(jù)傾斜會非常有幫助。4. 小練習(xí)高頻問題排查與避坑技巧4.1 報錯速查本地環(huán)境最容易踩的五個坑做練習(xí)的時候最打擊信心的事情就是代碼還沒跑起來先被環(huán)境折騰半天。我把自己踩過和幫別人看過的五個高頻報錯整理在下面。第一個是啟動 Python 時出現(xiàn)Unsupported class file major version這基本可以斷定是 JDK 版本和 Spark 不匹配。我的處理辦法是下載 JDK 17 并重新設(shè)置JAVA_HOME確保java -version輸出為 17。第二個是 Windows 下報Could not locate executable null\bin\winutils.exe這是本地缺少 Hadoop 運(yùn)行時導(dǎo)致的解決方法要么補(bǔ)齊對應(yīng)的依賴文件要么干脆用 WSL 或 Linux 環(huán)境。我自己的經(jīng)驗是為一個小練習(xí)去折騰本地 Hadoop 相關(guān)組件不太劃算能換環(huán)境就換環(huán)境。第三個問題很常見但經(jīng)常被忽視spark.read.csv讀不到文件。有人會把路徑寫成帶盤符的絕對路徑這在本地沒問題但如果你的代碼里包含中文路徑或特殊字符就會有一堆編碼問題。我更推薦把練習(xí)文件放在項目目錄下用相對路徑并確認(rèn)啟動 Python 的工作目錄在項目根目錄。第四個問題是寫 CSV 后中文亂碼一般發(fā)生在默認(rèn)編碼不是 UTF-8 的場景。你可以在寫出時加option(encoding, UTF-8)生成數(shù)據(jù)時也保證用 UTF-8 編碼。最后一個坑不是報錯而是結(jié)果對不上明明按月份分組但輸出的月份順序是亂序的因為 Spark 默認(rèn)不保證排序你需要顯式排序。4.2 一個常見誤區(qū)本地數(shù)據(jù)量小為什么還是慢我見過有人拿著這份小練習(xí)跑完一看發(fā)現(xiàn)很多 Stage 只有幾百毫秒但總時間卻要十幾秒于是懷疑 Spark 是不是“太重了”。其實這多半不是 Spark 的問題而是你沒有控制任務(wù)粒度。默認(rèn)情況下Spark SQL 的spark.sql.shuffle.partitions是 200意思是任何 Shuffle 操作都會分成 200 個分區(qū)去跑。你的數(shù)據(jù)才五萬行分成 200 個分區(qū)只會產(chǎn)生大量空任務(wù)和調(diào)度開銷。我在練習(xí)開始時把它改成了 8整個聚合任務(wù)明顯輕快不少。還有一個容易被忽略的點count()是一個 Action每次調(diào)用都會從源頭重新計算一遍。比如你在代碼里寫了df.count()又寫了df_clean.count()再運(yùn)行一次聚合前面幾步可能重復(fù)執(zhí)行了多次。如果數(shù)據(jù)量級稍微大一點這種重復(fù)成本會被放大。你可以用df.cache()把中間結(jié)果緩存起來不過緩存本身也有代價要放在多次復(fù)用的場景下才劃算。小練習(xí)里最值得學(xué)的不是堆緩存而是搞清楚哪些算子會觸發(fā) Shuffle、哪些操作是 Action。你只要把這兩點分清就不難理解“慢”到底慢在哪個環(huán)節(jié)。4.3 經(jīng)典問題對照表現(xiàn)象、原因、處理方案每次帶新人做這個練習(xí)我都會讓他們對照一張問題表排查效率比自己瞎猜高很多。這里整理成表格放在你旁邊對照即可現(xiàn)象常見原因處理建議啟動報Unsupported class file major versionJDK 版本超出 Spark 支持范圍使用 JDK 8/11/17重新設(shè)置 JAVA_HOMEWindows 下報 winutils.exe 找不到缺少 Hadoop Windows 本地依賴補(bǔ)齊依賴或改用 WSL/Linux 環(huán)境中文讀取或?qū)懗鰜y碼文件編碼不是 UTF-8讀寫時統(tǒng)一指定encoding文件本身也用 UTF-8 保存結(jié)果月份順序亂沒有顯式排序或排序字段類型是字符串對日期字段轉(zhuǎn)成 date 類型后再排序或加ORDER BY month數(shù)據(jù)特別小卻跑很慢默認(rèn) shuffle 分區(qū)數(shù)過大產(chǎn)生大量空任務(wù)設(shè)置spark.sql.shuffle.partitions為較小值聚合值比預(yù)期大沒過濾取消單、沒去重、空值未處理先做數(shù)據(jù)質(zhì)量檢查再進(jìn)入?yún)R總這張表只解決“能不能跑”的問題真正理解為什么還需要你回到前面的章節(jié)把每條現(xiàn)象背后的 Spark 原理串起來。比如為什么分區(qū)數(shù)影響速度為什么去重要放在聚合之前這些不是靠背而是靠親手改參數(shù)、觀察結(jié)果差異來獲得體感。5. 做完練習(xí)之后的幾條實在體會5.1 跑通只是第一步重點是看懂每一條轉(zhuǎn)換把這組練習(xí)完整跑完之后我最想說的一點是不要滿足于“代碼能出結(jié)果”。如果你的操作只是照著貼一遍那你的提升非常有限。Spark 的學(xué)習(xí)曲線不是靠記憶彎道超車而是靠對數(shù)據(jù)流轉(zhuǎn)的想象。每調(diào)用一個withColumn都會生成一條新的數(shù)據(jù)版本每執(zhí)行一次groupBy數(shù)據(jù)都會發(fā)生一次寬依賴也就是 Shuffle。你要在腦子里形成一條“數(shù)據(jù)從文件出發(fā)經(jīng)過哪幾步變成什么形狀最終落在哪里”的路徑。做完練習(xí)后我建議你把每一行代碼刪掉再重寫一遍不用看答案能寫出來才算真會了。這個復(fù)寫的過程比復(fù)制十遍都管用。5.2 同一個數(shù)據(jù)集還能怎么練出更多東西這個練習(xí)的數(shù)據(jù)集其實很有挖頭做完基礎(chǔ)版本之后你可以繼續(xù)加難度。比如用窗口函數(shù)計算每個品類按月銷售額的環(huán)比看哪些品類增長明顯也可以用用戶維度做一次關(guān)聯(lián)查詢把訂單表和用戶表關(guān)聯(lián)后分析復(fù)購率還可以把月結(jié)果和累計結(jié)果放在一起試一試ROLLUP或CUBE。這些擴(kuò)展仍然只用這一份模擬數(shù)據(jù)成本很低但每次都能把 Spark 的一個知識塊補(bǔ)得更完整。我自己做這個練習(xí)時額外加了一個讀取目錄下多個 CSV 文件的小任務(wù)本質(zhì)上和讀取單文件一樣但順帶理解了通配符路徑和分區(qū)發(fā)現(xiàn)的機(jī)制。等你把這些都試完再回頭看最初的小練習(xí)會發(fā)現(xiàn)它就是一個很好的 Spark 學(xué)習(xí)坐標(biāo)系。如果對你有幫助可以按我上面的步驟自己跑一遍尤其是問題排查那部分踩過一次坑記住的細(xì)節(jié)比刷十道題都牢。我最后再多說一句本地 Spark 小練習(xí)只是起點后續(xù)接上真實數(shù)據(jù)源、加上合理的分區(qū)策略、學(xué)會看執(zhí)行計劃你會發(fā)現(xiàn) Spark 那些看起來嚇人的概念其實都能在這次練習(xí)里找到落點。