您好,登錄后才能下訂單哦!
這篇文章主要介紹Spark-Sql的示例分析,文中介紹的非常詳細(xì),具有一定的參考價(jià)值,感興趣的小伙伴們一定要看完!
Spark SQL對SQL語句的處理,首先會將SQL語句進(jìn)行解析(Parse),然后形成一個(gè)Tree,在后續(xù)的如綁定、優(yōu)化等處理過程都是對Tree的操作,而操作的方法是采用Rule,通過模式匹配,對不同類型的節(jié)點(diǎn)采用不同的操作。
spark-sql是用來處理結(jié)構(gòu)化數(shù)據(jù)的模塊,是入門spark的首要模塊。
技術(shù)的學(xué)習(xí)無非就是去了解它的API,但是Spark有點(diǎn)難,因?yàn)樗睦雍途W(wǎng)上能搜到的基本都是Scala寫的。我們這里使用Java。
數(shù)據(jù)處理的第一個(gè)例子通常都是word count,就是統(tǒng)計(jì)一個(gè)文件里每個(gè)單詞出現(xiàn)了幾次。我們也來試一下。
> 這個(gè)例子網(wǎng)上有很多,即使是通過spark實(shí)現(xiàn)的也不少;這里面大部分都是使用Scala寫的,我沒有試過;少部分是通過Java寫的;
Java里面的例子有一些是使用RDD實(shí)現(xiàn)的,只有極個(gè)別是通過DataSet來做的。但即使這一小撮例子,我也跑不通。
所以我自己來嘗試完成這個(gè)例子,看到別人用Scala寫三五行就完成了,而我嘗試了一整天幾無進(jìn)展。在網(wǎng)上東拼西湊熟悉Spark的Java
還是以我們前面的例子來改:
String logFile = "words";
SparkSession spark = SparkSession.builder().appName("Simple Application").master("local").getOrCreate();
Dataset<String> logData = spark.read().textFile(logFile).cache();
System.out.println("行數(shù):" + logData.count());這里我不再使用之前的README文件,自己創(chuàng)建了一個(gè)words文件,內(nèi)容隨意寫了一堆單詞。
執(zhí)行程序,可以正常打印出來:
接下來我們需要把句子分割成一個(gè)個(gè)單詞合在一起,然后統(tǒng)計(jì)每個(gè)單詞出現(xiàn)的次數(shù)。
> 可能有人會說,這個(gè)簡單,我用Java8的流一下就處理好了:
把行集合通過flatMap處理,每一行通過split(" ")分割成一個(gè)獨(dú)立的單詞集合,再把結(jié)果通過自身groupBy一下就拿到終止數(shù)據(jù)結(jié)構(gòu)Map了。
最后把map的key和value的大小拿到就好了。
的確,使用Java就是這樣實(shí)現(xiàn)。但是Spark提供了一套和Java的流API名字和效果類似的工具,區(qū)別是Spark的是分布式API
我們通過Spark的flatMap先來處理一下:
Dataset<String> words = logData.flatMap((FlatMapFunction<String, String>) k -> Arrays.asList(k.split("\\s")).iterator(), Encoders.STRING()); System.out.println("單詞數(shù):" + words.count()); words.foreach(k -> { System.out.println("W:" + k); });
不同于Java的流,spark這個(gè)flatMap的返回值是可以直接訪問結(jié)果的:
> 可能有人留意到spark中函數(shù)式方法的參數(shù)定義和Java差距較大。他們的參數(shù)不太一樣,還多了個(gè)編碼器。目前來講我還不清楚為啥這樣定義,不過印象中編碼器也是spark3的重要優(yōu)化內(nèi)容。
再Java中使用Scala的方法總是有些怪異,Lambda表達(dá)式前面總是需要強(qiáng)制類型轉(zhuǎn)換,只是為了指明參數(shù)類型,否則需要new一個(gè)匿名類。
這個(gè)也花了我不少時(shí)間,后來找到一個(gè)網(wǎng)頁org.apache.spark.sql.Dataset.flatMap java code examples | Tabnine
再往后我迷茫了:
KeyValueGroupedDataset<String, String> group = words.groupByKey((Function1<String, String>) k -> k, Encoders.STRING());
這樣我已經(jīng)group好了,但是返回的不是DataSet,我也不知道這個(gè)返回有啥用,怎么拿到里面的內(nèi)容呢?我費(fèi)了好大勁沒搞定。
比如我發(fā)現(xiàn)count方法會返回一個(gè)DataSet:
看起來正是我想要的,但是當(dāng)我想把它輸出竟然執(zhí)行報(bào)錯(cuò):
ount.foreach(t -> { System.out.println(t); });
別說foreach了,就算想看看里面的數(shù)量(就像一開始我們查看了文件有幾行那樣)都會報(bào)錯(cuò),錯(cuò)誤內(nèi)容一樣
count.count();
查了很多資料,大意是說spark的計(jì)算方法都是分布式的,各個(gè)任務(wù)之間需要通信,通信時(shí)需要序列化來傳遞信息。所以上面我們能看文件行數(shù)因?yàn)轭愋褪荢tring,有序列化標(biāo)志;現(xiàn)在生成的是元組,不能序列化。我嘗試了各種方法,甚至自己創(chuàng)建新類模擬了計(jì)算過程還是不行
查了好久資料,比如Job aborted due to stage failure: Task not serializable: | Databricks Spark Knowledge Base (gitbooks.io)依然沒有解決。偶然的機(jī)會找到一個(gè)令人激動(dòng)的網(wǎng)站Spark Groupby Example with DataFrame — SparkByExamples終于解決了我的問題。
DataFrame雖然是spark提供的重要工具,但是再Java上并沒有對應(yīng)的類,只是把DataSet的泛型對象改成Row而已。注意這個(gè)Row沒有泛型定義,所以里面有哪些列不知道
可以從一開始就把DataSet轉(zhuǎn)成DataFrame:
但是可以看到要從Row里面拿數(shù)據(jù)比較麻煩。所以目前我只在需要序列化的地方轉(zhuǎn):
以上是“Spark-Sql的示例分析”這篇文章的所有內(nèi)容,感謝各位的閱讀!希望分享的內(nèi)容對大家有幫助,更多相關(guān)知識,歡迎關(guān)注億速云行業(yè)資訊頻道!
免責(zé)聲明:本站發(fā)布的內(nèi)容(圖片、視頻和文字)以原創(chuàng)、轉(zhuǎn)載和分享為主,文章觀點(diǎn)不代表本網(wǎng)站立場,如果涉及侵權(quán)請聯(lián)系站長郵箱:is@yisu.com進(jìn)行舉報(bào),并提供相關(guān)證據(jù),一經(jīng)查實(shí),將立刻刪除涉嫌侵權(quán)內(nèi)容。