柚子快報邀請碼778899分享：開發(fā)語言 scala學(xué)習(xí)

http://yzkb.51969.com/

目錄

1 常用變量數(shù)據(jù)類型

1.1 字符串輸出

1.2 控制臺輸入

1.3 文件的輸入輸出

1.4 scala的數(shù)據(jù)類型

Nothing：所有類型的子類

2 流程控制

2.1 for循環(huán)

for循環(huán)守衛(wèi)

for循環(huán)返回值

2.2 while循環(huán)

2.3 循環(huán)中斷

3 函數(shù)基礎(chǔ)，函數(shù)和方法

=>的四種用法：

3.2 閉包

3.3 惰性加載

4 面向?qū)ο?/p>

4.1 包對象

4.2 伴生對象

4.3 特質(zhì)trait

4.4 枚舉類和應(yīng)用類

5 集合

5.1 集合分類及介紹

5.2 常用集合方法

5.2.1 針對所有集合

5.2.2 針對seq或list

5.2.3 集合的計(jì)算函數(shù)

6 模式匹配

7 隱式轉(zhuǎn)換

8 泛型

9 正則表達(dá)式

10 補(bǔ)充

spark

spark core

Spark StructType

Struct Field

sparksql相關(guān)

scala中格式化字符串s和f的區(qū)別

spark sql中collect_list

groupBy

reducebyKey

agg函數(shù)

WrappedArray

創(chuàng)建臨時視圖

dataframe常用操作

UDF

lateral view explode()

CONCAT_WS(separator,str1,str2,...)

多行數(shù)據(jù)聚合為一行

mapPartition

字符串匹配

case when用法

生成唯一id

多線程

cache和persist

join操作

場景和做法

l 快速上手：可以參考B站scala快速入門視頻以及菜鳥教程的文字版。

官方文檔：Spark 3.5.1 ScalaDoc

學(xué)習(xí)視頻：B站尚硅谷。

本文總結(jié)基于B站尚硅谷視頻。

伴生類和伴生對象：因?yàn)閟cala取消了static關(guān)鍵字，所以全局唯一的屬性無法被創(chuàng)建，scala采用Object伴生對象來解決這個問題。例如scala類a向創(chuàng)建全局屬性b，但a中沒有方法可以做到，此時可以創(chuàng)建Object a伴生對象，在里面創(chuàng)建的屬性和方法可以被scala類a訪問（包括私有），同樣object伴生對象也能訪問所有的scala類a的方法和屬性（包括私有）。

伴生對象全局都有效且唯一。

1 常用變量數(shù)據(jù)類型

能用常量val盡量用val，因?yàn)榉虾瘮?shù)式編程的思想。

scala賦值時會自動根據(jù)賦值推斷類型，類型確定后不能再更改，變量聲明時必須賦初值。

字符串類型的*代表重復(fù)某個字符串多次。eg：”a”*3輸出3個a。

1.1 字符串輸出

輸出，字符串模板，s不能指定輸出格式，f可以指定個數(shù)。

雙引號就是輸出雙引號里面的內(nèi)容 三引號就是保持多行數(shù)據(jù)的原格式輸出

1.2 控制臺輸入

StdIn.方法

1.3 文件的輸入輸出

讀取文件內(nèi)的數(shù)據(jù)：Source.fromFile

寫數(shù)據(jù)到內(nèi)存里：可以直接調(diào)用java的IO方法

1.4 scala的數(shù)據(jù)類型

Nothing：所有類型的子類

意義：Scala的泛型是不允許不加類型參數(shù)的，比如說想新建list，新建時必須注明list的泛型，val list = List(1)是錯誤的，val list[Int] = List(1) 才是正確的。若需要用到空list，則可以用List[Nothing]。

Nil是繼承List[Nothing]的，當(dāng)要用到空list時直接用Nil就可以（經(jīng)常與集合的連接符號"::"連用）。

string類型的split(參數(shù)1)方法，參數(shù)1代表利用字符串的哪個符號進(jìn)行分割

注意點(diǎn)：若想利用|或.分割，需要變?yōu)閇|]或[.]

2 流程控制

2.1 for循環(huán)

單層for循環(huán)：

1 to 10，包含10 1 to 10 by 2，循環(huán)步長為2，每隔兩個取一個 1.0 to 10.0 by 0.5，循環(huán)步長為0.5，推薦用bigDecimal 1 until 10，不包含10 Range（start，end，step）

雙層for循環(huán)：

直接兩個for嵌套即可實(shí)現(xiàn) for(i <- 1 to 3;j <- 1 to 3)也可實(shí)現(xiàn)雙層for循環(huán)，等價于外層i，內(nèi)層j，共9次循環(huán)。

引入變量：

該變量和i相關(guān)，每個i對應(yīng)一個變量。

for(i <- 1 to 10;j = i+1) for{ i <- 1 to 10 j = i+1 }也可以實(shí)現(xiàn)，后面跟循環(huán)內(nèi)部邏輯即可

創(chuàng)建一個循環(huán)：

Seq.range(-11,-1).foreach

for循環(huán)守衛(wèi)

在for循環(huán)內(nèi)部循環(huán)范圍后面直接加if判斷語句，只取滿足if語句的值

for循環(huán)返回值

默認(rèn)情況下為空，unit。 val for(i <- 1 to 10) yield 要返回的變量名(eg:i或者i*i)，若for循環(huán)里放集合，則可以對每個元素都操作并創(chuàng)建返回值

2.2 while循環(huán)

循環(huán)前先判斷，沒有返回值，循環(huán)內(nèi)的變量需要先聲明。

2.3 循環(huán)中斷

用Breaks.breakable{}，內(nèi)包循環(huán)邏輯，循環(huán)邏輯要break的地方用Breaks.break()即可實(shí)現(xiàn)跳出循環(huán)

3 函數(shù)基礎(chǔ)，函數(shù)和方法

函數(shù)是寫在main方法內(nèi)的，方法是定義在類或?qū)ο罄锏?，調(diào)用時使用 對象.方法名 調(diào)用。 輸入?yún)?shù)可變：在類型后加"*"，可變參數(shù)一般放在最后。 輸入?yún)?shù)有默認(rèn)值：在類型后加"="，然后輸入默認(rèn)值即可。 指定變量的調(diào)用時賦值：在調(diào)用的括號內(nèi)寫變量名字然后賦值即可。 匿名函數(shù)：只關(guān)心處理邏輯，不關(guān)心名字。(變量名:變量類型)=>{方法體}

若匿名函數(shù)內(nèi)部變量只出現(xiàn)過一次，則變量名和變量類型和括號可以直接省略，然后用_代替每個變量 匿名函數(shù)作為參數(shù)傳給其他函數(shù)。 將函數(shù)作為整體傳遞給另一個常量，val f1=f _，注意f后有一個空格。

=>的四種用法：

表示函數(shù)的返回值類型 匿名函數(shù)的定義：左邊是參數(shù)，右邊是函數(shù)實(shí)現(xiàn)體 （x: Int）=>{} 模式匹配或try-catch：match 和 try-catch 都用 “=>” 表示輸出的結(jié)果或返回的值

4. 傳名參數(shù)：傳名參數(shù)在函數(shù)調(diào)用前表達(dá)式不會被求值，而是會作為一個匿名函數(shù)作為函數(shù)參數(shù)傳遞下去，例如參數(shù)類型為無參函數(shù)的參數(shù)就是傳名參數(shù)。

3.2 閉包

內(nèi)層函數(shù)用到了外層函數(shù)的局部變量。

使用場景：針對大部分?jǐn)?shù)據(jù)都沒改變的情況下進(jìn)行處理的場景，可以提高函數(shù)的通用性。

注意：

閉包是在每個executor上獨(dú)立執(zhí)行的，所以只能使用閉包外部的局部變量，但不會改變該外部的值 如果想獲得閉包內(nèi)部對某個值的修改情況，可以用accumulator累加器，最后輸出累加器的結(jié)果即可（注意一定要在action動作后執(zhí)行）

3.3 惰性加載

lazy關(guān)鍵字修飾，調(diào)用時才加載。

4 面向?qū)ο?/p>

對象轉(zhuǎn)換：Scala中isInstanceOf 和 asInstanceOf

scala的hashMap轉(zhuǎn)為java的hashMap：

首先新建類時用scala.collection.mutable.HashMap，然后只需在該map后面加asJava即可實(shí)現(xiàn)轉(zhuǎn)換，注意import scala.collection.JavaConverters.mutableMapAsJavaMapConverter

scala中用java得類，如果想用父類承接子類，會失敗報提示類型不匹配，解決辦法：

new 時候，如果能new父類，就new父類，如果父類是抽象類不能new，就new 子類，但要注意聲明得是父類。

4.1 包對象

package Object：里面的所有內(nèi)容在同一個包下的任何位置都可以訪問（必須是同一個包下的同一層級）。

4.2 伴生對象

可以訪問同名類的所有方法和屬性，包括私有的，同時伴生對象里的屬性和方法類似于靜態(tài)變量和方法，全局只有一個。

apply方法，針對類構(gòu)造器私有化的情況（在類名后面加private），無法直接new類，但可以在伴生對象中寫apply方法，接收參數(shù)并new類，并且在其他位置new類時不用顯示調(diào)用apply方法，直接new+類名就可以達(dá)到創(chuàng)建對象的效果。

unapply方法，將對象拆解為不同的屬性。

4.3 特質(zhì)trait

格式：extends A with B with... 可以在new后加with實(shí)現(xiàn)動態(tài)實(shí)現(xiàn)特質(zhì)，但注意要實(shí)現(xiàn)抽象方法。 特質(zhì)的疊加：若多個特質(zhì)有同名方法，則按照extends順序從后向前尋找super調(diào)用的方法，找到就停止尋找。 自身類型：解決依賴注入的問題。場景：某個類想訪問另一個類的屬性，但不想繼承另一個類，采用自身類型解決。 eg：UserDao特質(zhì)操作user類型，在UserDao里需要有一個User類型的變量，可以在trait中使用_: User=>，即可實(shí)現(xiàn)User的注入，可以采用this.name獲取User的name。

4.4 枚舉類和應(yīng)用類

枚舉類需要繼承Enumeration。

應(yīng)用類需繼承App。可以直接實(shí)現(xiàn)方法。

type：可以定義新類型。

5 集合

5.1 集合分類及介紹

分為seq序列，set集，map映射，都擴(kuò)展自iterable。

多維數(shù)組：Array.ofDim[泛型](幾維數(shù)組就寫幾個參數(shù)，最高5維)。

Nil：是一個空List

List（列表）：

插入和刪除比Array更高效，訪問元素一般是遍歷，其保存的值不能被更改。不能new創(chuàng)建，根據(jù)伴生對象的apply方法創(chuàng)建，即直接List（1,2,3,5,6）即可。 合并list操作：:::操作可以直接將一個list的所有元素加到另一個list中；::將前一個list作為整體和另一個list的所有元素合并。

ListBuffer（可變列表）：

創(chuàng)建：可new ListBuffer()創(chuàng)建，也可直接ListBuffer（1,2,3,5）創(chuàng)建 增：append方法；prepend（在前面添加）；insert（index，元素），在第index位置加多個元素；+=，在后面加；+=：按照位置從右到左加到原list的頭部。 合并兩個list：++，用法list1=list2++list3，此時創(chuàng)建了一個新列表，包含兩個list的所有元素；++=，用法list1++=list2，此時list1改變。 修改：list（index）=新元素；update（index，新元素），修改索引為index的元素為新元素。 刪除：remove（index），刪除索引為index的元素；-=元素a，刪除元素a

set，可變和不可變都同名，區(qū)別是引用的報名不同，所以set前要顯式聲明時可變還是不可變。

創(chuàng)建：set（1,2,4） 增：set + 新元素，不管是否是可變set，該方法都不會改變原set 合并：++操作 刪除：-=，直接刪除對應(yīng)元素

可變集合類型增：add（新元素）方法；+=方法 可變集合刪除：remove（刪除元素），返回Boolean。 可變集合合并：++=

Stream流

惰性集合，需要時才會加載。

示例：

一個list[Int]里有1,2,3,4,5，如果直接打印list，輸出1,2,3,4,5，但如果將list轉(zhuǎn)為stream s，然后打印s，會輸出1，？

方法：

tail，獲取除了第一個元素的后面元素集合 head，獲取第一個元素

map

創(chuàng)建：Map（key1->value1，key2->value2） 遍歷：map.foreach方法；map.keys：map所有的key；map.values：map的所有value； map.getOrElse()，獲取key對應(yīng)的value，若沒有就返回指定參數(shù)。

可變map，mutable.Map

添加：map.put（key，value）；map+=((key，value)) 更新：map.update(key,value)；+= 刪除：map.remove(key)；-=key 合并集合：++=

tuple元組，存放不同類型的集合

創(chuàng)建：直接(元素1，元素2)賦值即可，最多22個元素，元組可以嵌套。 查：._index，index從1開始；productElement(index)，index從0開始。 遍歷：productIterator方法，產(chǎn)生迭代器，用于遍歷。

queue隊(duì)列

創(chuàng)建：可變隊(duì)列可new，可用伴生類創(chuàng)建；不可變隊(duì)列只能用伴生類。 入隊(duì)：enqueue(元素1，元素2，...) 出隊(duì)：dequeue

并行集合

在集合后加.par即可實(shí)現(xiàn)并行。

獲取第一個元素：

.head()方法，但注意如果集合可能為null的話該方法會報錯，可用headOption.getOrElse()方法獲取第一個元素并賦默認(rèn)值

5.2 常用集合方法

5.2.1 針對所有集合

長度：length方法，seq分類的集合可以使用；size()方法都可以使用。 生成字符串：list和set自動實(shí)現(xiàn)了toString，可以直接打印對應(yīng)元素；也可以用mkString()自定義輸出格式。 是否包含：contains()方法。 翻轉(zhuǎn)：reverse方法。 取并集：union()方法，兩個集合的所有元素都合在一起，若是set會去重。 取交集：intersect()方法，取共同元素返回新集合。 取差集：list1.diff(list2)，返回1有的，2沒有的元素。 拉鏈：zip()方法，取兩個集合相同位置的元素組成一個二元組。 滑窗：sliding(n)方法，窗口大小為n，每次滑動一格；sliding(n，step)，窗口大小為n，每次滑step格。

5.2.2 針對seq或list

獲取對象頭：head方法；init方法，除掉最后一個元素的其他元素，返回一個新的集合。 獲取對象尾：tail方法只能獲取除頭部的其他元素，返回一個新的集合；last方法直接獲取最后一個元素。 獲取指定索引的元素（從零開始）：list(0)，list.apply(0)。 取前n個：take(n)方法，返回一個新集合。 取后n個：takeRight(n)，返回新集合 刪除前n個：drop(n)，返回新集合 刪除后n個：dropRight(n)，返回新集合 取最后一個元素：last

5.2.3 集合的計(jì)算函數(shù)

簡單函數(shù)

求和：sum方法，求集合所有元素的和 乘積：product 最大值：max；maxBy方法可以定制比較的是哪個參數(shù)，比如針對元組類型可以規(guī)定按第一個還是第二個或其他來比較。 最小值：min；minBy方法可以定制比較方法，同maxBy。 排序：sorted()方法，默認(rèn)升序，可以填入隱式參數(shù)Ordering；sortBy可以實(shí)現(xiàn)定制參數(shù)，同樣可以填入隱式參數(shù)實(shí)現(xiàn)升序或逆序排列；sortWith可以實(shí)現(xiàn)定制規(guī)則，傳入匿名函數(shù)。

高級函數(shù)：

過濾：遍歷集合將符合條件的元素放入到新集合中。filter方法，傳入匿名函數(shù)即可按照指定函數(shù)的規(guī)則選取。 map映射：將每個元素映射到某個函數(shù)。map方法，傳入匿名函數(shù)對每個元素進(jìn)行指定操作 flatten扁平化：若有嵌套集合，則取消集合的嵌套，將其數(shù)據(jù)結(jié)構(gòu)變?yōu)橐粚印? flatMap（扁平化+映射）：傳入匿名函數(shù)，即map對應(yīng)的操作，然后會對操作結(jié)果進(jìn)行扁平化處理。 groupBy分組：傳入匿名函數(shù)，得到的是map，按照指定規(guī)則對集合的元素進(jìn)行分組。 簡化，規(guī)約：reduce操作，傳入匿名函數(shù)，對flatMap后的數(shù)據(jù)操作，返回相同類型的數(shù)據(jù)，即由集合變成一個結(jié)果；reduceLeft，基本同reduce，但可以返回不同類型的數(shù)據(jù)，從左向右操作；reduceRight，從右向左操作，使用時需要注意，內(nèi)部是遞歸調(diào)用，邏輯比較復(fù)雜。 折疊：fold函數(shù)，傳入兩個參數(shù)，第一個是初始狀態(tài)，第二個是匿名函數(shù)，描述了對元素的操作，返回和初始狀態(tài)相同類型的數(shù)據(jù)，即由集合變成一個結(jié)果；foldLeft，基本同fold從左向右操作；foldRight，從右向左操作，使用時需要注意，內(nèi)部是先翻轉(zhuǎn)再調(diào)用foldLeft，邏輯比較復(fù)雜。

問題：返回的集合是.fold前的集合還是新的集合？ 錯，返回的不是集合，是一個元素，是傳入的初始化狀態(tài)。

reduce用法：

val dates: List[String] = List("2020-01-01", "2020-01-02", "2020-01-03")

val rootPath: String = _

//讀取日志文件，去重、并展開userInterestList

def createDF(rootPath: String, date: String): DataFrame = {

val path: String = rootPath + date

val df = spark.read.parquet(path)

.distinct

.withColumn("userInterest", explode(col("userInterestList")))

df

}

//提取字段、過濾，再次去重，把多天的結(jié)果用union合并

val distinctItems: DataFrame = dates.map{

case date: String =>

val df: DataFrame = createDF(rootPath, date)

.select("userId", "itemId", "userInterest", "accessFreq")

.filter("accessFreq in ('High', 'Medium')")

.distinct

df

}

.reduce(_ union _)

6 模式匹配

match-case，必須要要有case_。

模式守衛(wèi)，匹配某個范圍：case后加if語句。

for (ch <- "+-3){

val sign = 0

val digital = 1

match ch {

case '+' => sign = 1

case '-' => sign = -1

case _ if ch.toString.equals("3") => digital = 3

case _ => sign = 2

}

println ("result = " + (sign*digital))

}

注意：若想用模式匹配去匹配類型，要注意scala的泛型擦除機(jī)制，比如List[String]無法識別里面的泛型，只能接收到是個List。 列表匹配,下面程序的結(jié)果是first=1,second=2,rest=List(3,5)，因?yàn)?:是列表相加的操作。

對象匹配：需要實(shí)現(xiàn)對象的unapply方法；或者直接用樣例類匹配，就不用專門去實(shí)現(xiàn)apply和unapply了。 模式中的變量：如果case后跟著定義了一個變量，會將match前的變量值賦給case后跟的變量 類型匹配：可以匹配變量類型，進(jìn)行不同操作 一個case中可以有多個條件，用"|"，"&&"操作符判斷是或還是與 val env = "st";

env match {

case "prod" =>

println("prod")

case "staging" | "st" =>

println("st")

case _ =>

println("others")

}

7 隱式轉(zhuǎn)換

實(shí)現(xiàn)將某個類型轉(zhuǎn)換為指定類型調(diào)用指定類型方法的功能，有需要時再關(guān)注。

8 泛型

9 正則表達(dá)式

Scala中提供了 Regex類 來定義正則表達(dá)式. 要構(gòu)造一個Regex對象，直接使用 String類的r方法 即可. 建議使用三個雙引號來表示正則表達(dá)式，不然就得對正則中的反斜杠進(jìn)行轉(zhuǎn)義.

調(diào)用方式："""表達(dá)式內(nèi)容""".r

10 補(bǔ)充

scala的try-catch不能捕獲assert

spark

spark中生成rdd的操作是轉(zhuǎn)換操作，計(jì)算操作（action）會對rdd進(jìn)行處理。

在Spark程序中，如果要在分布式計(jì)算過程中使用某個對象，那么這個對象必須能夠被序列化，以便將其傳輸?shù)郊褐械钠渌?jié)點(diǎn)，若該類中存在不可被序列化的對象，運(yùn)行時會報錯：java.io.NotSerializableException

如何解決：

確保它們都實(shí)現(xiàn)了Serializable接口 如果某個類中包含了不可序列化的成員變量，可以將這些成員變量標(biāo)記為@transient，這樣Spark就不會嘗試序列化它們了。注意：這樣做會導(dǎo)致類反序列化時將這些變量設(shè)為初始值，可能會導(dǎo)致值丟失的情況。

可以被序列化的：

object類型的單例對象 對象中的方法（若方法中用了不可被序列化的變量，則一樣會報錯）

不可被序列化的：

listbuffer類型的變量

spark core

Spark StructType

用于創(chuàng)建dataframe和dateSet

是一個seq，里面放著多個Struct Field

StructType 是個case class,一般用于構(gòu)建schema. 因?yàn)槭莄ase class,所以使用的時候可以不用new關(guān)鍵字

里面包的是Struct Field

Struct Field

結(jié)構(gòu)如下：

case class StructField( name: String,

dataType: DataType,

nullable: Boolean = true,

metadata: Metadata = Metadata.empty) {}

sparksql相關(guān)

scala中格式化字符串s和f的區(qū)別

三引號括起來的情況下

寫spark.sql語句時，f""""""中若想比較字符串類型的字段是否等于某個值a，值a必須用雙引號括起來，單引號會報錯。

寫spark.sql語句時，s""""""中若想比較字符串類型的字段是否等于某個值a，值a用單雙引號都可以。

spark sql中collect_list

將多行數(shù)據(jù)合并為一行的list。

hive或者spark中collect_list一般是用來做分組后的合并。也可以和partition by連用。

groupBy

傳入匿名函數(shù)。比如contains，模式匹配，或者按照指定要求分類。只能分組，不能聚合。若聚合需要再加map操作。

用法一：對多個列排序，同時計(jì)數(shù)，起別名

df.groupby(['Column1','Column2']).agg(count("*").alias("count"), avg("Column3").alias("Column4")).show()

用法二：groupby后排序

df.groupby(['column1','columns2']).agg(count("*").alias("count"), avg("column3").alias("column4")).orderBy(['count','column4'],ascending=[0,1])

注意：

groupby可以用null作為分組的一個key，spark sql會將所有null值看作一個新鍵放在一起

reducebyKey

會對分區(qū)內(nèi)的數(shù)據(jù)進(jìn)行預(yù)聚合，可以在shuffle之前對分區(qū)內(nèi)相同key的數(shù)據(jù)集進(jìn)行預(yù)聚合（combine）功能，這樣會減少落盤的數(shù)據(jù)量，而groupByKey只是進(jìn)行分組，不存在數(shù)據(jù)量減少的問題，reduceByKey性能比較高。

注意：reduceByKey只能針對rdd進(jìn)行，rdd需要先轉(zhuǎn)換為（k，v），接著進(jìn)行reduceByKey操作，該操作會在每個分區(qū)內(nèi)先聚合相同key為一行數(shù)據(jù)，然后再聚合不同分區(qū)的數(shù)據(jù)。

agg函數(shù)

在整體DataFrame不分組聚合，里面可以傳入聚合函數(shù)，可用于groupBy后聚合非groupBy的字段。

示例：

WrappedArray

常用于獲取collect_list后的結(jié)果，格式XXX.getAs[mutable.WrappedArray[T]]。是Array的包裝類，功能比Array多。

方法列表：

indices，獲取索引列表，在遍歷時使用。

創(chuàng)建臨時視圖

一個sparkSession生命周期內(nèi)的臨時視圖，可以被看做是一個數(shù)據(jù)源，但不會被持久化，可以使用sparksql語句查詢里面的內(nèi)容，由dataframe創(chuàng)建臨時視圖：

df.createOrReplaceTempView("my_temp_view")

也可以創(chuàng)建跨sparkSession的全局臨時視圖，使用createOrReplaceGlobalTempView方法，此時生命周期被綁在spark應(yīng)用程序的生命周期，而不是單個spark-session。注意：全局臨時視圖保存在系統(tǒng)保留的數(shù)據(jù)庫global_temp中，查詢時需要加global_temp前綴。

dataframe常用操作

【Spark學(xué)習(xí)筆記】 Scala DataFrame操作大全-CSDN博客

getAs[T](fieldIndex(fieldName))：獲取dataframe指定index的數(shù)據(jù)，從右向左排序，索引從0開始？ 為某個可能為null的字段賦初始值，有兩種方法（when otherwise和coalesce）：在處理空字段時，coalesce函數(shù)更適合用于填充默認(rèn)值，而when函數(shù)更適合用于根據(jù)條件進(jìn)行選擇。 val defaultArray = typedLit(Seq.empty[Double])

df.withColumn("start_index_old", when(col("start_index_old").isNull, (defaultArray)).otherwise(col("start_index_old")))，注意該方法可能報錯類型不一致，此時可以使用printSchema函數(shù)來查看該列的數(shù)據(jù)類型，以確定正確的返回值類型。 df.withColumn("start_index_old", coalesce(col("start_index_old"), defaultArray)) 創(chuàng)建空dataframe：

定義數(shù)據(jù)結(jié)構(gòu)：StructType結(jié)構(gòu)，該結(jié)構(gòu)代表一個dataframe或dateset中的一行數(shù)據(jù)。里面有多個StructField，每個StructField中有三個字段，第一個字段表示列名（string），第二個字段代表列的類型（IntegerType），第三個字段代表是否可空（true或false） 創(chuàng)建dataframe，ss.createDataFrame(ss.sparkContext.emptyRDD[Row], scheme) import org.apache.spark.sql.types._ 合并dataframe，union操作：相同元素會去重。unionAll：相同元素不去重。

UDF

（User Define Function 用戶自定義函數(shù)），Spark SQL擴(kuò)展接口用于定義新的基于列的函數(shù)，這些函數(shù)擴(kuò)展了Spark SQL的DSL用于轉(zhuǎn)換數(shù)據(jù)集的詞匯表。對于dataframe類型，可以將某列的數(shù)據(jù)進(jìn)行用戶自定義的一些操作。

用法：

定義一個函數(shù)字面量（Function Literal），即java中的匿名函數(shù)類，賦值給一個變量。 將變量作為入?yún)鹘oudf函數(shù)，將其保存為一個用戶自定義函數(shù)。

//將字符串轉(zhuǎn)為seq[string]，同時替換里面的雙引號

val parseToList: String => Seq[String] = _.trim.replaceAll("\"", "").stripPrefix("[").stripSuffix("]").split(",").toSeq.map(_.toString)

val parseToListUDF = udf(parseToList)

//將字符串轉(zhuǎn)為seq[long]

val parseToLongList: String => Seq[Long] = _.trim.stripPrefix("[").stripSuffix("]").split(",").toSeq.map(_.toLong)

val parseToLongListUDF = udf(parseToLongList)

//將多個json體構(gòu)成的list，拆成多個string類型的json體，用seq包起來

val splitJsonObjects: String => Seq[String] = { str =>

val pattern: Regex = "\\{[^\\{\\}]*\\}".r

pattern.findAllIn(str).toList

}

// 注冊 UDF

val parseToBraceListUDF = udf(splitJsonObjects)

lateral view explode()

用于一行轉(zhuǎn)多行：

SQL 之 lateral view explode()_sql lateral view explore-CSDN博客

用法：

explode函數(shù)將分割后的數(shù)組轉(zhuǎn)為多行，每行包含數(shù)組中的一個元素，需要與lateral一起使用，實(shí)現(xiàn)將原始表的每行與explode生成的行組合

魔數(shù)：

SELECT entity_id, geometry, property, geometry_status, field_source_id, exploded.source_id

FROM (

SELECT entity_id, geometry, property, geometry_status,

get_json_object(property, '$.field_source_id') as field_source_id,

get_json_object(property, '$.source_id') as source_id_json

FROM app_mapdata.master_data_current_entity_feature_prod

WHERE changeset_lastest_id = 386306541

AND get_json_object(property, '$.source_id') IS NOT NULL

and get_json_object(property,'$.source_id')!=''

AND entity_id = 14014353169

) md

LATERAL VIEW explode(split(md.source_id_json, '|')) exploded AS source_id

sparksql：

方式1：和魔數(shù)的用法一致，但要注意特殊字符|用[|]代替

方式2：利用sparksql的內(nèi)置函數(shù)，可以直接跟select使用

SELECT entity_id, geometry, property, geometry_status, field_source_id, explode(split(source_id_json, '[|]')) as source_id

FROM (

SELECT entity_id, geometry, property, geometry_status,

get_json_object(property, '$$.field_source_id') as field_source_id,

get_json_object(property, '$$.source_id') as source_id_json

FROM $SOURCE_FEATURE_ENTITY

WHERE changeset_lastest_id = $CHANGESET_LASTEST_ID_ENTITY

AND get_json_object(property, '$$.source_id') IS NOT NULL

and get_json_object(property,'$$.source_id')!=''

) md

CONCAT_WS(separator,str1,str2,...)

連接字符串：

concat_ws("_", field1, field2)，輸出結(jié)果將會是：“field1_field2”。

連接數(shù)組元素：

concat_ws("_", [a,b,c])，輸出結(jié)果將會是："a_b_c"。

多行數(shù)據(jù)聚合為一行

collect_set：把聚合的數(shù)據(jù)組合成一個數(shù)組，一般和group_by使用，具有去重的效果。聚組后分割符是逗號 collect_list：把聚合的數(shù)據(jù)組合成一個數(shù)組，一般和group_by使用，不具有去重的效果。聚組后分割符是逗號

mapPartition

與map類似，map是對rdd的每一個元素進(jìn)行操作，而mapPartition（foreachPartition）是對rdd每個分區(qū)的迭代器進(jìn)行操作。

特點(diǎn)：

每個分區(qū)會應(yīng)用一次函數(shù)，所以函數(shù)的調(diào)用次數(shù)少，速度更快。分區(qū)內(nèi)的數(shù)據(jù)會變成一個迭代器傳進(jìn)去，函數(shù)的返回值也是迭代器 一個分區(qū)的數(shù)據(jù)會被同時加載到內(nèi)存中，可能會引發(fā)內(nèi)存不足的問題

注意：輸入是迭代器，輸出也是迭代器（iterator）

如果想要獲取迭代器的大小，一定要將其轉(zhuǎn)為list后再獲取，如果直接獲取會有問題（迭代器被消耗完，后面就是對空的迭代器進(jìn)行操作）

使用場景：

在map過程中需要頻繁創(chuàng)建額外的對象（比如將rdd的數(shù)據(jù)通過jdbc寫入數(shù)據(jù)庫，map需要為每個元素創(chuàng)建一個鏈接，而mappartition是為每個partition創(chuàng)建一個鏈接）

使用：

傳入Iterator迭代器，返回Iterator類型，對應(yīng)泛型可以不同

mapPartition計(jì)算效率比map高：

def testMapPartition(ss : SparkSession):

Unit ={

val sc = ss.sparkContext

val rdd = sc.parallelize(1 to 10000000) // 創(chuàng)建一個包含 1 到 10 的 RDD

// 使用 map

val begin = System.currentTimeMillis()

val mapResult = rdd.map(x => x * 2)

val mapResultList = mapResult.collect().mkString(",")

println("process time = %d s".format((System.currentTimeMillis() - begin)/1000))

// println(mapResultList)

// 使用 mapPartitions

val begin1 = System.currentTimeMillis()

val mapPartitionsResult = rdd.mapPartitions { iter =>

iter.map(x => x * 2) // 在迭代器上應(yīng)用 map 操作

}

val mapPartitionsResultList = mapPartitionsResult.collect().mkString(",")

println("process time = %d s".format((System.currentTimeMillis() - begin1)/1000))

// println(mapPartitionsResultList)

}

結(jié)果：

process time = 1 s（map）

process time = 0 s（mapPartition）

字符串匹配

直接在sql語句中寫xxx like ('匹配表達(dá)式')，spark中存在危險，可能會匹配和表達(dá)式不同的內(nèi)容 sql語句選取完內(nèi)容后加一個flatmap，在里面確定待匹配項(xiàng)，利用scala函數(shù)來進(jìn)行字符串匹配 dataframe的filter(col("列名").like("表達(dá)式")).select("想要的列名")

建議采用：方法三，dataframe的filter(col("列名").like("表達(dá)式")).select("想要的列名")

case when用法

30. Spark SQL case when用法：_sparksql case when-CSDN博客

示例：將下列sql轉(zhuǎn)為sparksql

SELECT c.PROCESS_ID,

CASE WHEN c.PAYMODE = 'M'

THEN

CASE WHEN CURRENCY = 'USD'

THEN c.PREMIUM * c.RATE

ELSE c.PREMIUM END * 12

ELSE

CASE WHEN CURRENCY = 'USD'

THEN c.PREMIUM * c.RATE

ELSE c.PREMIUM END END VAlue

FROM CMM c

生成唯一id

monotonically_increasing_id?是 Spark SQL 中的一個函數(shù)，它生成一個唯一的、遞增的長整型 ID 值。這個 ID 對于每一行數(shù)據(jù)都是唯一的，并且在整個數(shù)據(jù)集中是遞增的。但是，這個 ID 并不是連續(xù)的。

這個函數(shù)在處理大數(shù)據(jù)集時非常有用，因?yàn)樗梢詾槊恳恍袛?shù)據(jù)生成一個唯一的 ID，而不需要進(jìn)行 shuffle 操作。這使得它在處理大規(guī)模數(shù)據(jù)時非常高效。

多線程

其實(shí)spark是支持在一個spark context中可以通過多線程同時提交多個任務(wù)運(yùn)行，然后spark context接到這所有的任務(wù)之后，通過中央調(diào)度，在來分配執(zhí)行各個task，最終任務(wù)完成程序退出。

推薦使用Executors.newFixedThreadPool來創(chuàng)建多線程。

cache和persist

RDD的cache()方法其實(shí)調(diào)用的就是persist方法，緩存策略均為MEMORY_ONLY； 可以通過persist方法手工設(shè)定StorageLevel來滿足工程需要的存儲級別； cache或者persist并不是action； 附：cache和persist都可以用unpersist來取消

考慮內(nèi)存消耗問題，倘若我們要處理的數(shù)據(jù)僅僅是進(jìn)行一次處理，用完即丟棄，就應(yīng)該避免使用cache或persist，從而降低對內(nèi)存的損耗。若確實(shí)需要將數(shù)據(jù)加載到內(nèi)存中，而內(nèi)存又不足以加載，則可以設(shè)置Storage Level。0.9版本的Spark提供了三種Storage Level：MEMORY_ONLY（這是默認(rèn)值），MEMORY_AND_DISK，以及DISK_ONLY。

join操作

Spark 中支持多種連接類型：

Inner Join : 內(nèi)連接；

Full Outer Join : 全外連接；

Left Outer Join : 左外連接；

Right Outer Join : 右外連接；

Left Semi Join : 左半連接，等價于RDS的in

Left Anti Join : 左反連接，等價于RDS的not in

Natural Join : 自然連接；

Cross (or Cartesian) Join : 交叉 (或笛卡爾) 連接。

join操作的連接條件：

兩個dataframe有相同列名，可以用seq(列名1，列名2)實(shí)現(xiàn)join 兩個dataframe列名不同，可以用df1(列名1)+布爾計(jì)算符+df2（列名2）的方式進(jìn)行join

場景和做法

將兩個數(shù)據(jù)組成一個n元組：struct 利用struct(列名1，列名2,...)的辦法實(shí)現(xiàn)

柚子快報邀請碼778899分享：開發(fā)語言 scala學(xué)習(xí)

http://yzkb.51969.com/

文章鏈接