1. 項(xiàng)目概述從零到一掌握Scala核心集合與WordCount如果你正在學(xué)習(xí)大數(shù)據(jù)開發(fā)尤其是Spark那么Scala這門語言是你繞不開的坎。很多朋友一上來就被函數(shù)式編程、不可變集合這些概念搞得頭大更別提去理解那些眼花繚亂的算子比如map、flatMap、reduce了。我剛開始接觸時(shí)也踩過不少坑比如分不清List和Set的應(yīng)用場景搞不懂sortBy和sortWith到底該用哪個(gè)寫個(gè)最簡單的WordCount都磕磕絆絆。今天我就結(jié)合自己踩過的這些“坑”把Scala里最常用、也最容易混淆的集合類型——元組Tuple、列表List、集合Set——以及排序和并行處理掰開揉碎了講清楚。我們不止講語法更重點(diǎn)講“為什么”要這么用以及在大數(shù)據(jù)場景下比如Spark它們是如何發(fā)揮作用的。最后我們會(huì)手把手實(shí)現(xiàn)一個(gè)經(jīng)典的WordCount案例并深入探討幾種不同的排序方法。目標(biāo)是讓你看完后不僅能寫出代碼更能理解背后的設(shè)計(jì)思想在面對復(fù)雜數(shù)據(jù)處理任務(wù)時(shí)能清晰地選擇最合適的工具。2. 核心數(shù)據(jù)結(jié)構(gòu)深度解析Tuple, List, Set2.1 元組Tuple輕量級的數(shù)據(jù)“包裹”元組在Scala里就像一個(gè)固定大小、可以存放不同類型數(shù)據(jù)的“包裹”。它最大的特點(diǎn)是輕量和快速。當(dāng)你需要臨時(shí)將幾個(gè)值捆綁在一起傳遞但又覺得專門定義一個(gè)case class有點(diǎn)小題大做時(shí)元組就是最佳選擇。2.1.1 元組的創(chuàng)建與訪問Scala支持創(chuàng)建最多22個(gè)元素的元組這個(gè)限制源于Scala與JVM的兼容性設(shè)計(jì)實(shí)際中極少用到這么多。// 創(chuàng)建元組 val tuple1 (1, “hello”) // 類型為(Int, String) val tuple2 “Alice” - 25 // 另一種創(chuàng)建方式等價(jià)于(“Alice”, 25)常用于創(chuàng)建鍵值對 val tuple3 (1, “data”, true, 3.14) // 混合類型元組 // 訪問元素使用從1開始的下劃線加序號(hào) println(tuple1._1) // 輸出1 println(tuple1._2) // 輸出hello注意訪問元素的下標(biāo)是從_1開始的而不是_0。這是Scala從數(shù)學(xué)和函數(shù)式語言如Haskell繼承來的習(xí)慣剛開始很容易寫錯(cuò)。2.1.2 元組解構(gòu)Pattern Matching逐個(gè)用_1、_2訪問很麻煩更優(yōu)雅的方式是解構(gòu)val (id, name) tuple1 println(s”ID: $id, Name: $name”) // 輸出ID: 1, Name: hello // 在match-case中尤其強(qiáng)大 tuple3 match { case (_, desc, flag, _) if flag println(s”$desc is active”) case _ println(“unknown”) }2.1.3 元組在大數(shù)據(jù)中的應(yīng)用場景在Spark中元組無處不在。例如reduceByKey算子處理后的數(shù)據(jù)就是(Key, Value)形式的元組RDD。它的輕量級特性使得在數(shù)據(jù)Shuffle混洗和網(wǎng)絡(luò)傳輸時(shí)效率極高。當(dāng)你需要對數(shù)據(jù)進(jìn)行簡單的聚合或轉(zhuǎn)換并產(chǎn)生一個(gè)臨時(shí)的、結(jié)構(gòu)簡單的復(fù)合結(jié)果時(shí)首先考慮元組。2.2 列表List不可變序列的基石Scala的List是一個(gè)不可變的、鏈表結(jié)構(gòu)的序列。這意味著一旦創(chuàng)建其內(nèi)容就不能改變?nèi)魏涡薷牟僮魅缣砑印h除都會(huì)返回一個(gè)全新的List。這聽起來可能低效但在函數(shù)式編程和大數(shù)據(jù)并行計(jì)算中不可變性避免了共享狀態(tài)下的并發(fā)修改問題是安全性的重要保障。2.2.1 列表的創(chuàng)建與基本操作// 創(chuàng)建列表 val list1 List(1, 2, 3, 4, 5) val list2 1 :: 2 :: 3 :: Nil // 使用::cons操作符從頭部構(gòu)建Nil代表空列表 // 基本操作 val head list1.head // 獲取第一個(gè)元素1 val tail list1.tail // 獲取除第一個(gè)元素外的列表List(2,3,4,5) val isEmpty list1.isEmpty // false // 連接列表 val list3 list1 ::: List(6, 7) // List(1,2,3,4,5,6,7)實(shí)操心得對空列表調(diào)用.head或.tail會(huì)拋出NoSuchElementException。安全的做法是先用isEmpty判斷或者使用headOption方法返回Option[T]。2.2.2 列表的高階方法Higher-Order Methods這是Scala集合的精髓。高階方法指的是以函數(shù)作為參數(shù)的方法它們讓你能以聲明式的方式描述“做什么”而不是“怎么做”。map一對一轉(zhuǎn)換遍歷列表對每個(gè)元素應(yīng)用一個(gè)函數(shù)生成一個(gè)新列表。val doubled list1.map(x x * 2) // List(2, 4, 6, 8, 10) val stringified list1.map(_.toString) // List(“1”, “2”, “3”, “4”, “5”)map是Spark RDD和DataFrame中最核心的轉(zhuǎn)換算子之一用于數(shù)據(jù)清洗和格式轉(zhuǎn)換。flatMap先map再“壓平”flatten它接收一個(gè)返回集合的函數(shù)然后將所有結(jié)果連接成一個(gè)列表。val sentences List(“Hello world”, “Hello Scala”) val words sentences.flatMap(_.split(“ “)) // List(Hello, world, Hello, Scala) // 等價(jià)于sentences.map(_.split(“ “)).flatten在WordCount中我們正是用flatMap將一行文本拆分成多個(gè)單詞。filter過濾保留滿足謂詞返回Boolean的函數(shù)的元素。val evens list1.filter(_ % 2 0) // List(2, 4)reduce與fold聚合reduce使用一個(gè)二元操作符將列表中的元素從左到右合并。val sum list1.reduce((a, b) a b) // 15 val sumShort list1.reduce(_ _) // 等價(jià)寫法注意reduce要求列表非空否則會(huì)拋異常。reduceLeft和reduceRight可以指定結(jié)合方向。fold與reduce類似但允許指定一個(gè)初始值zero value。這解決了空列表的問題并且初始值的類型可以和元素類型不同。val totalChars List(“a”, “bb”, “ccc”).fold(0)((acc, str) acc str.length) // 6在Spark中reduce和aggregate更通用的fold是行動(dòng)算子Action會(huì)觸發(fā)實(shí)際計(jì)算。foreach遍歷執(zhí)行副作用對每個(gè)元素應(yīng)用一個(gè)返回Unit的函數(shù)常用于打印日志或更新外部變量。list1.foreach(println)2.2.3 列表的不可變性與性能考量因?yàn)椴豢勺兠看巍靶薷摹倍紩?huì)創(chuàng)建新對象。頻繁在列表頭部添加元素是高效的::操作時(shí)間復(fù)雜度O(1)但在尾部添加則需要遍歷整個(gè)列表O(n)。如果需要頻繁的尾部追加應(yīng)考慮使用Vector不可變隨機(jī)訪問和修改都接近O(1)或ListBuffer可變最后可轉(zhuǎn)換為List。2.3 集合Set無序且唯一的容器Set存儲(chǔ)唯一的、無序的元素。它基于哈希表HashSet或紅黑樹TreeSet實(shí)現(xiàn)檢查元素是否存在contains的速度非常快。2.3.1 集合的創(chuàng)建與操作// 創(chuàng)建集合 val set1 Set(1, 2, 3, 2, 1) // Set(1, 2, 3)重復(fù)元素被去重 val emptySet Set.empty[String] // 基本操作 set1.contains(2) // true set1 4 // 添加元素返回新SetSet(1, 2, 3, 4) set1 - 2 // 移除元素返回新SetSet(1, 3) set1 Set(3, 4, 5) // 并集Set(1, 2, 3, 4, 5) set1 Set(2, 3, 4) // 交集Set(2, 3)2.3.2 可變Set與不可變Set默認(rèn)導(dǎo)入的scala.collection.immutable.Set是不可變的。如果需要原地修改使用scala.collection.mutable.HashSet。import scala.collection.mutable.HashSet val mutableSet HashSet(1, 2, 3) mutableSet 4 // 原地添加mutableSet變?yōu)镾et(1, 2, 3, 4)在大數(shù)據(jù)開發(fā)中除非有非常明確的性能優(yōu)化需求如在單機(jī)內(nèi)存中頻繁更新一個(gè)去重集合否則優(yōu)先使用不可變集合以避免并發(fā)環(huán)境下的數(shù)據(jù)競爭問題。2.3.3 Set在大數(shù)據(jù)中的應(yīng)用在Spark中Set常被用作廣播變量Broadcast Variable。例如你有一個(gè)需要過濾的敏感詞列表這個(gè)列表較小但會(huì)被所有任務(wù)節(jié)點(diǎn)頻繁讀取。將其作為一個(gè)Set廣播出去每個(gè)節(jié)點(diǎn)本地存一份副本可以極大減少網(wǎng)絡(luò)IO和序列化開銷。val stopWords sc.broadcast(Set(“the”, “a”, “an”, “in”, “on”)) val filteredRDD textRDD.filter(word !stopWords.value.contains(word))3. 經(jīng)典案例實(shí)戰(zhàn)從多角度實(shí)現(xiàn)WordCountWordCount是大數(shù)據(jù)領(lǐng)域的“Hello World”。我們通過它來串聯(lián)前面講到的集合和高階方法。假設(shè)我們有一個(gè)文本行組成的列表。3.1 基礎(chǔ)版分步拆解理解過程val lines List(“hello world hello scala”, “hello spark from scala”) // 1. 拆分將每行字符串拆分成單詞數(shù)組然后壓平成一個(gè)所有單詞的列表 val words lines.flatMap(_.split(“ “)) // List(hello, world, hello, scala, hello, spark, from, scala) // 2. 映射將每個(gè)單詞轉(zhuǎn)換成(單詞, 1)的元組形式為計(jì)數(shù)做準(zhǔn)備 val wordPairs words.map(word (word, 1)) // List((hello,1), (world,1), (hello,1), (scala,1), (hello,1), (spark,1), (from,1), (scala,1)) // 3. 分組按照單詞元組的第一個(gè)元素進(jìn)行分組 val grouped wordPairs.groupBy(_._1) // Map(hello - List((hello,1), (hello,1), (hello,1)), world - List((world,1)), …) // 4. 統(tǒng)計(jì)對分組后的每個(gè)列表計(jì)算其長度即1的個(gè)數(shù) val wordCounts grouped.map { case (word, list) (word, list.size) } // Map(hello - 3, world - 1, scala - 2, spark - 1, from - 1) // 5. 轉(zhuǎn)換為列表并輸出 wordCounts.toList.sortBy(-_._2).foreach(println) // 輸出(hello,3) (scala,2) (world,1) (spark,1) (from,1)這個(gè)版本清晰地展示了WordCount的每一步flatMap-map-groupBy-map。但它有一個(gè)性能問題groupBy會(huì)在內(nèi)存中構(gòu)建一個(gè)巨大的Map[List[...]]如果數(shù)據(jù)量極大可能導(dǎo)致內(nèi)存溢出。3.2 優(yōu)化版使用groupBy與mapValues上述步驟3和4可以合并利用mapValues直接對分組后的值列表進(jìn)行操作val wordCounts lines .flatMap(_.split(“ “)) .groupBy(identity) // 按單詞本身分組得到 Map[String, List[String]] .mapValues(_.size) // 直接對每個(gè)List求長度identity是一個(gè)函數(shù)輸入什么就返回什么x x。這里更簡潔但groupBy的內(nèi)存問題依然存在。3.3 生產(chǎn)常用版模擬reduceByKey邏輯Spark RDD的reduceByKey是解決WordCount的標(biāo)準(zhǔn)且高效的方法因?yàn)樗鼤?huì)在Map端先進(jìn)行本地合并Combine大大減少Shuffle的數(shù)據(jù)量。我們在Scala集合上模擬這個(gè)思想val wordCounts lines .flatMap(_.split(“ “)) .map(word (word, 1)) .groupBy(_._1) // 分組得到Map(word - List((word,1), (word,1)...)) .map { case (word, tuples) (word, tuples.map(_._2).sum) // 對每個(gè)分組內(nèi)的所有1求和 } // 或者使用更函數(shù)式的foldLeft // .map { case (word, tuples) (word, tuples.foldLeft(0)(_ _._2)) }雖然這里最終還是用了groupBy但在Spark分布式環(huán)境下reduceByKey的“映射-歸約”模型能極大提升性能。理解這個(gè)單機(jī)模擬版對理解Spark算子的工作原理至關(guān)重要。4. 并行集合Parallel Collections與排序4.1 并行處理數(shù)據(jù)利用多核能力對于計(jì)算密集型的操作Scala標(biāo)準(zhǔn)庫提供了并行集合可以自動(dòng)將任務(wù)分配到多個(gè)CPU核心上執(zhí)行語法非常簡單只需在集合后加上.par。val list (1 to 1000000).toList // 串行計(jì)算 val serialSum list.filter(_ % 2 0).map(_ * 2).sum // 并行計(jì)算 val parallelSum list.par.filter(_ % 2 0).map(_ * 2).sum只需要將list換成list.par后續(xù)的filter、map、sum操作就會(huì)并行執(zhí)行。4.1.1 并行化的注意事項(xiàng)開銷并行化本身有開銷任務(wù)切分、線程調(diào)度、結(jié)果合并對于小數(shù)據(jù)量比如少于1000個(gè)元素串行往往更快。副作用在并行操作中絕對要避免修改外部可變變量。因?yàn)閳?zhí)行順序是不確定的會(huì)導(dǎo)致數(shù)據(jù)競爭和不可預(yù)知的結(jié)果。// 錯(cuò)誤示范 var sum 0 list.par.foreach(sum _) // sum的結(jié)果每次運(yùn)行都可能不同順序并行集合上的操作如map不保證結(jié)果元素的順序與原始集合一致。如果需要保持順序可以在并行計(jì)算后調(diào)用.seq轉(zhuǎn)回順序集合但可能會(huì)損失部分性能。4.1.2 何時(shí)使用并行集合適用于數(shù)據(jù)量較大、每個(gè)元素處理成本較高、且操作是無狀態(tài)和無副作用的純函數(shù)場景。例如大規(guī)模數(shù)值計(jì)算、圖像批量處理等。在大數(shù)據(jù)領(lǐng)域真正的并行計(jì)算由Spark、Flink等框架在集群級別處理單機(jī)并行集合可作為補(bǔ)充。4.2 排序三劍客sorted, sortBy, sortWith的區(qū)別Scala提供了三種主要的排序方法它們各有側(cè)重。4.2.1sorted基于自然排序要求集合元素必須實(shí)現(xiàn)scala.math.Ordered特質(zhì)或Java的Comparable接口。對于Int、String、Tuple按元素依次比較等已有默認(rèn)排序。val nums List(3, 1, 4, 1, 5) nums.sorted // List(1, 1, 3, 4, 5) val words List(“banana”, “apple”, “cherry”) words.sorted // List(apple, banana, cherry) (按字典序)4.2.2sortBy指定排序依據(jù)接收一個(gè)函數(shù)將元素映射到一個(gè)具有自然排序類型的值通常是Int、String等然后根據(jù)這個(gè)映射后的值進(jìn)行排序。這是最常用的排序方法。case class Person(name: String, age: Int) val people List(Person(“Alice”, 25), Person(“Bob”, 20), Person(“Charlie”, 25)) // 按年齡排序 people.sortBy(_.age) // List(Bob(20), Alice(25), Charlie(25)) // 按年齡降序然后按姓名升序 people.sortBy(p (-p.age, p.name)) // List(Alice(25), Charlie(25), Bob(20))sortBy非常靈活和高效因?yàn)樗恍枰?jì)算一次映射函數(shù)然后對映射結(jié)果排序。4.2.3sortWith自定義比較邏輯接收一個(gè)比較函數(shù)(A, A) Boolean。當(dāng)函數(shù)返回true時(shí)表示第一個(gè)參數(shù)應(yīng)排在第二個(gè)參數(shù)之前。它提供了最強(qiáng)大也是最底層的控制。// 按字符串長度排序 words.sortWith(_.length _.length) // List(apple, banana, cherry) 假設(shè)長度不同 // 復(fù)雜的自定義排序年齡大的在前年齡相同則名字短的在前 people.sortWith { (a, b) if (a.age ! b.age) a.age b.age else a.name.length b.name.length }4.2.4 對比與選型指南特性sortedsortBysortWith排序依據(jù)元素自身的自然順序元素某個(gè)屬性的自然順序自定義的兩兩比較邏輯性能高直接比較高計(jì)算一次鍵值相對較低多次比較靈活性低中高使用場景對內(nèi)置類型數(shù)字、字符串或已定義Ordered的自定義類進(jìn)行標(biāo)準(zhǔn)排序。最常用。根據(jù)一個(gè)或多個(gè)可比較的字段進(jìn)行排序。需要復(fù)雜、非標(biāo)準(zhǔn)的排序邏輯時(shí)。示例List(5,2,8).sortedpeople.sortBy(_.age)people.sortWith((a,b) a.age b.age)選型建議絕大多數(shù)情況下使用sortBy。它語法簡潔意圖明確且性能優(yōu)異。只有當(dāng)排序邏輯無法通過簡單映射到某個(gè)字段來表達(dá)時(shí)例如需要根據(jù)多個(gè)字段進(jìn)行有條件的復(fù)雜比較才使用sortWith。5. 常見問題與性能調(diào)優(yōu)實(shí)錄在實(shí)際開發(fā)中僅僅知道語法是不夠的如何高效、正確地使用這些工具才是關(guān)鍵。下面分享幾個(gè)我踩過的坑和總結(jié)的經(jīng)驗(yàn)。5.1 集合選擇不當(dāng)導(dǎo)致性能瓶頸問題在一個(gè)需要頻繁隨機(jī)訪問和更新的數(shù)據(jù)緩存場景錯(cuò)誤地使用了List導(dǎo)致程序性能極差。分析List的隨機(jī)訪問時(shí)間復(fù)雜度是O(n)。對于需要按索引頻繁查找的場景應(yīng)使用Array可變、快速隨機(jī)訪問或Vector不可變、良好的隨機(jī)訪問和更新性能。排查技巧遇到性能問題首先分析集合的主要操作增、刪、查、改及其頻率。對照下表進(jìn)行選擇操作ListVectorArray(可變)Set(HashSet)頭部訪問/添加O(1)O(log32(n))O(1)N/A隨機(jī)訪問O(n)O(log32(n))O(1)O(1) (平均)查找元素O(n)O(n)O(n)O(1)(平均)尾部添加O(n)O(log32(n))O(n) (需擴(kuò)容)O(1) (平均)5.2 在并行操作或閉包中修改外部變量問題為了統(tǒng)計(jì)par.foreach中的處理數(shù)量使用了一個(gè)外部AtomicInteger雖然線程安全但造成了嚴(yán)重的鎖競爭并行效率反而不如串行。解決優(yōu)先使用無副作用的轉(zhuǎn)換和聚合操作。例如用par.map和reduce代替foreach和外部計(jì)數(shù)器。// 不佳 val count new java.util.concurrent.atomic.AtomicInteger(0) bigList.par.foreach { _ count.incrementAndGet() } // 更佳 val count bigList.par.map(_ 1).sum5.3groupBy的內(nèi)存溢出OOM風(fēng)險(xiǎn)場景在單機(jī)處理一個(gè)非常大的數(shù)據(jù)集時(shí)直接使用groupBy對某個(gè)字段進(jìn)行分組。風(fēng)險(xiǎn)groupBy會(huì)在內(nèi)存中構(gòu)建一個(gè)Map[K, List[V]]如果鍵K的數(shù)量非常多或者每個(gè)鍵對應(yīng)的列表List[V]非常大極易導(dǎo)致OOM。規(guī)避策略考慮使用groupBy的惰性視圖groupBy有一個(gè)lazy版本groupByLazy在某些Scala版本中但它只是延遲了中間結(jié)果的物化最終仍需要內(nèi)存。分流處理如果可能將數(shù)據(jù)分批處理或者先使用filter過濾掉不需要的數(shù)據(jù)。使用aggregate或fold對于可結(jié)合associative和可交換commutative的聚合操作如求和、求最大值可以遍歷一次數(shù)據(jù)使用Map[K, V]來累加結(jié)果避免存儲(chǔ)整個(gè)列表。這模擬了Spark中reduceByKey的思想。val data List((“a”, 1), (“b”, 2), (“a”, 3)) val result data.foldLeft(Map.empty[String, Int]) { (acc, kv) acc (kv._1 - (acc.getOrElse(kv._1, 0) kv._2)) }終極方案對于真正的大數(shù)據(jù)應(yīng)使用Spark等分布式計(jì)算框架讓數(shù)據(jù)在集群中分布存儲(chǔ)和計(jì)算。5.4sortBy與sortWith的穩(wěn)定性問題排序后相等元素的原始相對順序是否保留答案Scala的默認(rèn)排序算法是穩(wěn)定的。這意味著如果兩個(gè)元素根據(jù)排序鍵被認(rèn)為是相等的例如sortBy(_.age)年齡相同那么它們在排序后的列表中的相對順序會(huì)與排序前保持一致。這在需要多次排序例如先按部門排再按薪資排時(shí)非常有用。sortWith的穩(wěn)定性取決于你提供的比較函數(shù)是否定義了相等的概念。