藉助真實案例和程式碼樣本，本文作者展示瞭如何將Sparke和MySQL結合起來，創造資料分析上的強大工具。

Apache Spark是一個類似Apache Hadoop的叢集計算框架，在Wikipedia上有大量描述：Apache Spark是一個開源叢集計算框架，出自加州大學伯克利分校的AMPLab，後被捐贈給了Apache軟體基金會。

相對於Hadoop基於磁碟的兩段式MapReduce規範，Spark基於記憶體的多段式基元在特定應用上表現要優出100倍。Spark允許使用者程式將資料載入到叢集記憶體中反覆查詢，非常適合機器學習演算法。

與流行的看法相反，Spark不需要將所有資料存入記憶體，但會使用快取來加速操作（就像MySQL那樣）。Spark也能獨立執行而無需Hadoop，並可以執行在單獨一臺伺服器上（甚至筆記本或桌上型電腦上），並充分利用所有CPU核心。開啟它並使用分散式模式真的很簡單。先開啟master，在同一個節點上執行slave：

然後在任何額外的節點上執行Spark worker（確定向/etc/hosts 添加了hostname或者使用DNS）：

在很多工中MySQL（開箱即用的）表現並不太好。MySQL的限制之一在於：1次查詢=1個CPU核心。也就是說，即便你有48個速度飛快的核心，外加一個大型資料集可用，也無法充分利用所有的計算能力，相反Spark卻能充分使用CPU核心。

MySQL與Spark的另一差異在於：

l MySQL使用所謂的“寫時模式（schema on write）”——需要將資料轉化到MySQL中，如果未包含在MySQL裡，就無法使用sql來查詢。

l Spark（還有Hadoop/Hive）使用“讀時模式（schema on read）”——比如在一個壓縮txt檔案頂部使用表格結構（或者其他支援的輸入格式），將其看作表格；然後我們可以用SQL來查詢這個“表格”。

也就是說，MySQL負責儲存+處理，而Spark只負責處理，並可直接貫通資料與外部資料集（Hadoop、Amazon S3，本地檔案、JDBC MySQL或其他資料集）的通道。Spark支援txt檔案（壓縮的）、SequenceFile、其他Hadoop輸入格式和Parquet列式儲存。相對Hadoop來說，Spark在這方面更為靈活：例如Spark可以直接從MySQL中讀取資料。

向MySQL載入外部資料的典型管道（pipeline）是：

1、 解壓縮（尤其是壓縮成txt檔案的外部資料）；

2、用“LOAD DATA INFILE”命令將其載入到MySQL的儲存表格中；

3、只有這樣，我們才能篩選/進行分組，並將結果儲存到另一張表格中。

這會導致額外的開銷；在很多情況下，我們不需要“原始”資料，但仍需將其載入MySQL中。

相反，我們的分析結果（比如聚合資料）應當存在MySQL中。將分析結果存在MySQL中並非必要，不過更為方便。假設你想要分析一個大資料集（即每年的銷售額對比），需要使用表格或圖表的形式展現出來。由於會進行聚合，結果集將會小很多，將其存在MySQL中與很多標準程式一同協作處理將會容易許多。

一個有趣的免費資料集是Wikipedia的頁數（從2008年啟用後到現在，壓縮後大於1TB）。這個資料可以下載（壓縮空間確定txt檔案），在AWS上也是可用的（有限資料集）。資料以小時聚合，包括以下欄位：

l專案（比如en，fr等，通常是一種語言）

l頁頭（uri），採用urlencode編碼

l請求數

l返回內容的大小

（資料欄位編譯到了檔名中，每小時1個檔案）

我們的目標是：找出英文版wiki中每日請求數位居前10的頁面，不過還要支援對任意詞的搜尋，方便闡釋分析原理。例如，將2008到2015年間關於“Myspace”和“Facebook”的文章請求數進行對比。使用MySQL的話，需要將其原封不動的載入MySQL。所有檔案按內建的日期編碼分佈。解壓的總大小大於10TB。下面是可選的步驟方案（典型的MySQL方式）：

1、解壓檔案並執行“LOAD DATA INFILE”命令，將其錄入臨時表格：

2、“插入到”最終的表格，進行聚合：

3、通過url解碼標題（可能用到UDF）。

開銷很大：解壓並將資料轉化為MySQL格式，絕大部分都會被丟棄，作為損耗。

根據我的統計，整理6年來的資料需耗時超過1個月，還不包括解壓時間，隨著表格逐漸變大、索引需要更新所帶來的載入時間折損。當然，有很多辦法可以加速這一過程，比如載入不同的MySQL例項、首先載入記憶體表格再集合成InnoDB等。

不過最簡單的辦法是使用Apache Spark和Python指令碼（pyspark）。Pyspark可以讀出原始的壓縮txt檔案，用SQL進行查詢，使用篩選、類似urldecode函式等，按日期分組，然後將結果集儲存到MySQL中。

下面是執行操作的Python指令碼：

在指令碼中用到了Spark來讀取原始壓縮檔案（每次一天）。我們可以使用目錄作為“輸入”或者檔案列表。然後用彈性分散式資料集（RDD）轉化格式；Python包含lambda函式對映和篩選，允許我們將“輸入檔案”分離並進行篩選。

下一步是應用模式（declare fields）；我們還能使用其他函式，比如用urllib.unquote來解碼標題（urldecode）。最終，我們可以註冊臨時表格，然後使用熟悉的SQL來完成分組。

該指令碼可以充分利用CPU核心。此外，即便不使用Hadoop，在分散式環境中執行也非常簡易：只要將檔案複製到SparkNFS/外部儲存。

該指令碼花了1個小時，使用了三個box，來處理一個月的資料，並將聚合資料載入到MySQL上（單一例項）。我們可以估出：載入全部6年的（聚合）資料到MySQL上需要大約3天左右。

你可能會問，為什麼現在要快得多（而且實現了同樣的例項）。答案是：管道不同了，而且更為有效。在我們起初的MySQL管道中，載入的是原始資料，需要大約數月時間完成。而在本案例中，我們在讀取時篩選、分組，然後只將需要的內容寫入MySQL。

這裡還有一個問題：我們真的需要整個“管道”嗎？是否可以簡單地在“原始”資料之上執行分析查詢？答案是：確實有可能，但是也許會需要1000個節點的Spark叢集才能奏效，因為需要掃描的資料量高達5TB（參見下文中的“補充”）。

通過使用group_res.write.jdbc(url=mysql_url, table=”wikistats.wikistats_by_day_spark”, mode=”append”) ，Spark會啟動多執行緒插入。

Spark提供了web介面，方便對工作進行監控管理。樣例如下：執行wikistats.py application：

結果：使用Parquet分列格式與MySQL InnoDB表格

Spark支援Apache Parquet分列格式，因此我們可以將RDD儲存為parquet檔案（存入HDFS時可以儲存到目錄中）：

我們將管道結果（聚合資料）存入Spark。這次使用了按天分割槽（“mydate=20080101”），Spark可以在這種格式中自動發現分割槽。得到結果後要進行查詢。假設我們想要找到2018年1月查詢最頻繁的10大wiki頁面。可以用MySQL進行查詢（需要去掉主頁和搜尋頁）：

請注意，我們已經使用了聚合（資料彙總）表格，而不是“原始”資料。我們可以看到，查詢花了1小時22分鐘。由於將同樣的結果存入了Parquet（見指令碼）中，現在可以在Spark-SQL中使用它了：

這將用到spark-sql的本地版本，而且只用到1個主機。

耗時大約20分鐘，比之前更快。

Apache Spark是分析和聚合資料的好辦法，而且非常簡便。我喜歡Spark與其他大資料還有分析框架的原因在於：

l開源與積極開發

l不依賴工具，例如輸入資料與輸出資料不一定非要藉助Hadoop

l獨立模式，啟動迅速，易於部署

l大規模並行，易於新增節點

l支援多種輸入與輸出格式；比如可以讀取/寫入MySQL（Java資料庫連線驅動）與Parquet分列格式

但是，也有很多缺點：

l技術太新，會有一些bug和非法行為。很多錯誤難以解釋。

l需要Java；Spark 1.5僅支援Java 7及以上版本。這也意味著需要額外記憶體——合情合理。

l你需要通過“spark-submit”來執行任務。

我認為作為工具，Apache Spark十分不錯，補足了MySQL在資料分析與商業智慧方面的短板。

【編輯推薦】