Spark-Sql入門程序示例詳解

Posted on by

SparkSQL運行架構

Spark SQL對SQL語句的處理，首先會將SQL語句進行解析（Parse），然後形成一個Tree，在後續的如綁定、優化等處理過程都是對Tree的操作，而操作的方法是采用Rule，通過模式匹配，對不同類型的節點采用不同的操作。

spark-sql是用來處理結構化數據的模塊，是入門spark的首要模塊。

技術的學習無非就是去瞭解它的API，但是Spark有點難，因為它的例子和網上能搜到的基本都是Scala寫的。我們這裡使用Java。

入門例子

數據處理的第一個例子通常都是word count，就是統計一個文件裡每個單詞出現瞭幾次。我們也來試一下。

> 這個例子網上有很多，即使是通過spark實現的也不少；這裡面大部分都是使用Scala寫的，我沒有試過；少部分是通過Java寫的；

Java裡面的例子有一些是使用RDD實現的，隻有極個別是通過DataSet來做的。但即使這一小撮例子，我也跑不通。

所以我自己來嘗試完成這個例子，看到別人用Scala寫三五行就完成瞭，而我嘗試瞭一整天幾無進展。在網上東拼西湊熟悉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(“行數：” + logData.count());這裡我不再使用之前的README文件，自己創建瞭一個words文件，內容隨意寫瞭一堆單詞。

執行程序，可以正常打印出來：

接下來我們需要把句子分割成一個個單詞合在一起，然後統計每個單詞出現的次數。

> 可能有人會說，這個簡單，我用Java8的流一下就處理好瞭：

把行集合通過flatMap處理，每一行通過split(” “)分割成一個獨立的單詞集合，再把結果通過自身groupBy一下就拿到終止數據結構Map瞭。

最後把map的key和value的大小拿到就好瞭。

的確，使用Java就是這樣實現。但是Spark提供瞭一套和Java的流API名字和效果類似的工具，區別是Spark的是分佈式API

我們通過Spark的flatMap先來處理一下：

Dataset<String> words = logData.flatMap((FlatMapFunction<String, String>) k -> Arrays.asList(k.split("\\s")).iterator(), Encoders.STRING());
System.out.println("單詞數：" + words.count());
words.foreach(k -> {
System.out.println("W:" + k);
});

不同於Java的流，spark這個flatMap的返回值是可以直接訪問結果的：

> 可能有人留意到spark中函數式方法的參數定義和Java差距較大。他們的參數不太一樣，還多瞭個編碼器。目前來講我還不清楚為啥這樣定義，不過印象中編碼器也是spark3的重要優化內容。

再Java中使用Scala的方法總是有些怪異，Lambda表達式前面總是需要強制類型轉換，隻是為瞭指明參數類型，否則需要new一個匿名類。

這個也花瞭我不少時間，後來找到一個網頁org.apache.spark.sql.Dataset.flatMap java code examples | Tabnine

再往後我迷茫瞭：

KeyValueGroupedDataset<String, String> group = words.groupByKey((Function1<String, String>) k -> k, Encoders.STRING());

這樣我已經group好瞭，但是返回的不是DataSet，我也不知道這個返回有啥用，怎麼拿到裡面的內容呢？我費瞭好大勁沒搞定。

比如我發現count方法會返回一個DataSet：

看起來正是我想要的，但是當我想把它輸出竟然執行報錯：

ount.foreach(t -> {
    System.out.println(t);
});

別說foreach瞭，就算想看看裡面的數量（就像一開始我們查看瞭文件有幾行那樣）都會報錯，錯誤內容一樣

count.count();

查瞭很多資料，大意是說spark的計算方法都是分佈式的，各個任務之間需要通信，通信時需要序列化來傳遞信息。所以上面我們能看文件行數因為類型是String，有序列化標志；現在生成的是元組，不能序列化。我嘗試瞭各種方法，甚至自己創建新類模擬瞭計算過程還是不行

查瞭好久資料，比如Job aborted due to stage failure: Task not serializable: | Databricks Spark Knowledge Base (gitbooks.io)依然沒有解決。偶然的機會找到一個令人激動的網站Spark Groupby Example with DataFrame — SparkByExamples終於解決瞭我的問題。

使用DataFrame

DataFrame雖然是spark提供的重要工具，但是再Java上並沒有對應的類，隻是把DataSet的泛型對象改成Row而已。註意這個Row沒有泛型定義，所以裡面有哪些列不知道

可以從一開始就把DataSet轉成DataFrame:

但是可以看到要從Row裡面拿數據比較麻煩。所以目前我隻在需要序列化的地方轉：

到此這篇關於Spark-Sql入門程序的文章就介紹到這瞭,更多相關Spark-Sql入門內容請搜索WalkonNet以前的文章或繼續瀏覽下面的相關文章希望大傢以後多多支持WalkonNet！

推薦閱讀：