如何使用Apache Spark和MySQL實(shí)現(xiàn)數(shù)據(jù)分析

本篇內(nèi)容主要講解“如何使用Apache Spark和MySQL實(shí)現(xiàn)數(shù)據(jù)分析”，感興趣的朋友不妨來(lái)看看。本文介紹的方法操作簡(jiǎn)單快捷，實(shí)用性強(qiáng)。下面就讓小編來(lái)帶大家學(xué)習(xí)“如何使用Apache Spark和MySQL實(shí)現(xiàn)數(shù)據(jù)分析”吧!

與流行的看法相反，Spark不需要將所有數(shù)據(jù)存入內(nèi)存，但會(huì)使用緩存來(lái)加速操作（就像MySQL那樣）。Spark也能獨(dú)立運(yùn)行而無(wú)需Hadoop，并可以運(yùn)行在單獨(dú)一臺(tái)服務(wù)器上（甚至筆記本或臺(tái)式機(jī)上），并充分利用所有CPU內(nèi)核。開(kāi)啟它并使用分布式模式真的很簡(jiǎn)單。先打開(kāi)master，在同一個(gè)節(jié)點(diǎn)上運(yùn)行slave：

然后在任何額外的節(jié)點(diǎn)上運(yùn)行Spark worker（確定向/etc/hosts 添加了hostname或者使用DNS）：

在很多任務(wù)中MySQL（開(kāi)箱即用的）表現(xiàn)并不太好。MySQL的限制之一在于：1次查詢(xún)=1個(gè)CPU內(nèi)核。也就是說(shuō)，即便你有48個(gè)速度飛快的內(nèi)核，外加一個(gè)大型數(shù)據(jù)集可用，也無(wú)法充分利用所有的計(jì)算能力，相反Spark卻能充分使用CPU內(nèi)核。

MySQL與Spark的另一差異在于：

l  MySQL使用所謂的“寫(xiě)時(shí)模式（schema on write）”——需要將數(shù)據(jù)轉(zhuǎn)化到MySQL中，如果未包含在MySQL里，就無(wú)法使用sql來(lái)查詢(xún)。

l  Spark（還有Hadoop/Hive）使用“讀時(shí)模式（schema on read）”——比如在一個(gè)壓縮txt文件頂部使用表格結(jié)構(gòu)（或者其他支持的輸入格式），將其看作表格；然后我們可以用SQL來(lái)查詢(xún)這個(gè)“表格”。

也就是說(shuō)，MySQL負(fù)責(zé)存儲(chǔ)+處理，而Spark只負(fù)責(zé)處理，并可直接貫通數(shù)據(jù)與外部數(shù)據(jù)集（Hadoop、Amazon S3，本地文件、JDBC MySQL或其他數(shù)據(jù)集）的通道。Spark支持txt文件（壓縮的）、SequenceFile、其他Hadoop輸入格式和Parquet列式存儲(chǔ)。相對(duì)Hadoop來(lái)說(shuō)，Spark在這方面更為靈活：例如Spark可以直接從MySQL中讀取數(shù)據(jù)。

向MySQL加載外部數(shù)據(jù)的典型管道（pipeline）是：

1、 解壓縮（尤其是壓縮成txt文件的外部數(shù)據(jù)）；

2、用“LOAD DATA INFILE”命令將其加載到MySQL的存儲(chǔ)表格中；

3、只有這樣，我們才能篩選/進(jìn)行分組，并將結(jié)果保存到另一張表格中。

這會(huì)導(dǎo)致額外的開(kāi)銷(xiāo)；在很多情況下，我們不需要“原始”數(shù)據(jù)，但仍需將其載入MySQL中。

相反，我們的分析結(jié)果（比如聚合數(shù)據(jù)）應(yīng)當(dāng)存在MySQL中。將分析結(jié)果存在MySQL中并非必要，不過(guò)更為方便。假設(shè)你想要分析一個(gè)大數(shù)據(jù)集（即每年的銷(xiāo)售額對(duì)比），需要使用表格或圖表的形式展現(xiàn)出來(lái)。由于會(huì)進(jìn)行聚合，結(jié)果集將會(huì)小很多，將其存在MySQL中與很多標(biāo)準(zhǔn)程序一同協(xié)作處理將會(huì)容易許多。

一個(gè)有趣的免費(fèi)數(shù)據(jù)集是Wikipedia的頁(yè)數(shù)（從2008年啟用后到現(xiàn)在，壓縮后大于1TB）。這個(gè)數(shù)據(jù)可以下載（壓縮空間確定txt文件），在AWS上也是可用的（有限數(shù)據(jù)集）。數(shù)據(jù)以小時(shí)聚合，包括以下字段：

l項(xiàng)目（比如en，fr等，通常是一種語(yǔ)言）

l頁(yè)頭（uri），采用urlencode編碼

l請(qǐng)求數(shù)

l返回內(nèi)容的大小

（數(shù)據(jù)字段編譯到了文件名中，每小時(shí)1個(gè)文件）

我們的目標(biāo)是：找出英文版wiki中每日請(qǐng)求數(shù)位居前10的頁(yè)面，不過(guò)還要支持對(duì)任意詞的搜索，方便闡釋分析原理。例如，將2008到2015年間關(guān)于“Myspace”和“Facebook”的文章請(qǐng)求數(shù)進(jìn)行對(duì)比。使用MySQL的話，需要將其原封不動(dòng)的載入MySQL。所有文件按內(nèi)置的日期編碼分布。解壓的總大小大于10TB。下面是可選的步驟方案（典型的MySQL方式）：

1、解壓文件并運(yùn)行“LOAD DATA INFILE”命令，將其錄入臨時(shí)表格：

2、“插入到”最終的表格，進(jìn)行聚合：

3、通過(guò)url解碼標(biāo)題（可能用到UDF）。

開(kāi)銷(xiāo)很大：解壓并將數(shù)據(jù)轉(zhuǎn)化為MySQL格式，絕大部分都會(huì)被丟棄，作為損耗。

根據(jù)我的統(tǒng)計(jì)，整理6年來(lái)的數(shù)據(jù)需耗時(shí)超過(guò)1個(gè)月，還不包括解壓時(shí)間，隨著表格逐漸變大、索引需要更新所帶來(lái)的加載時(shí)間折損。當(dāng)然，有很多辦法可以加速這一過(guò)程，比如載入不同的MySQL實(shí)例、首先載入內(nèi)存表格再集合成InnoDB等。

不過(guò)最簡(jiǎn)單的辦法是使用Apache Spark和Python腳本（pyspark）。Pyspark可以讀出原始的壓縮txt文件，用SQL進(jìn)行查詢(xún)，使用篩選、類(lèi)似urldecode函數(shù)等，按日期分組，然后將結(jié)果集保存到MySQL中。

下面是執(zhí)行操作的Python腳本：

在腳本中用到了Spark來(lái)讀取原始?jí)嚎s文件（每次一天）。我們可以使用目錄作為“輸入”或者文件列表。然后用彈性分布式數(shù)據(jù)集（RDD）轉(zhuǎn)化格式；Python包含lambda函數(shù)映射和篩選，允許我們將“輸入文件”分離并進(jìn)行篩選。

下一步是應(yīng)用模式（declare fields）；我們還能使用其他函數(shù)，比如用urllib.unquote來(lái)解碼標(biāo)題（urldecode）。最終，我們可以注冊(cè)臨時(shí)表格，然后使用熟悉的SQL來(lái)完成分組。

該腳本可以充分利用CPU內(nèi)核。此外，即便不使用Hadoop，在分布式環(huán)境中運(yùn)行也非常簡(jiǎn)易：只要將文件復(fù)制到SparkNFS/外部存儲(chǔ)。

該腳本花了1個(gè)小時(shí)，使用了三個(gè)box，來(lái)處理一個(gè)月的數(shù)據(jù)，并將聚合數(shù)據(jù)加載到MySQL上（單一實(shí)例）。我們可以估出：加載全部6年的（聚合）數(shù)據(jù)到MySQL上需要大約3天左右。

你可能會(huì)問(wèn)，為什么現(xiàn)在要快得多（而且實(shí)現(xiàn)了同樣的實(shí)例）。答案是：管道不同了，而且更為有效。在我們起初的MySQL管道中，載入的是原始數(shù)據(jù)，需要大約數(shù)月時(shí)間完成。而在本案例中，我們?cè)谧x取時(shí)篩選、分組，然后只將需要的內(nèi)容寫(xiě)入MySQL。

這里還有一個(gè)問(wèn)題：我們真的需要整個(gè)“管道”嗎？是否可以簡(jiǎn)單地在“原始”數(shù)據(jù)之上運(yùn)行分析查詢(xún)？答案是：確實(shí)有可能，但是也許會(huì)需要1000個(gè)節(jié)點(diǎn)的Spark集群才能奏效，因?yàn)樾枰獟呙璧臄?shù)據(jù)量高達(dá)5TB（參見(jiàn)下文中的“補(bǔ)充”）。

通過(guò)使用group_res.write.jdbc(url=mysql_url, table=”wikistats.wikistats_by_day_spark”, mode=”append”) ，Spark會(huì)啟動(dòng)多線程插入。

Spark提供了web接口，方便對(duì)工作進(jìn)行監(jiān)控管理。樣例如下：運(yùn)行wikistats.py application：

結(jié)果：使用Parquet分列格式與MySQL InnoDB表格

Spark支持Apache Parquet分列格式，因此我們可以將RDD存儲(chǔ)為parquet文件（存入HDFS時(shí)可以保存到目錄中）：

我們將管道結(jié)果（聚合數(shù)據(jù)）存入Spark。這次使用了按天分區(qū)（“mydate=20080101”），Spark可以在這種格式中自動(dòng)發(fā)現(xiàn)分區(qū)。得到結(jié)果后要進(jìn)行查詢(xún)。假設(shè)我們想要找到2018年1月查詢(xún)最頻繁的10大wiki頁(yè)面。可以用MySQL進(jìn)行查詢(xún)（需要去掉主頁(yè)和搜索頁(yè)）：

請(qǐng)注意，我們已經(jīng)使用了聚合（數(shù)據(jù)匯總）表格，而不是“原始”數(shù)據(jù)。我們可以看到，查詢(xún)花了1小時(shí)22分鐘。由于將同樣的結(jié)果存入了Parquet（見(jiàn)腳本）中，現(xiàn)在可以在Spark-SQL中使用它了：

這將用到spark-sql的本地版本，而且只用到1個(gè)主機(jī)。

耗時(shí)大約20分鐘，比之前更快。

Apache Spark是分析和聚合數(shù)據(jù)的好辦法，而且非常簡(jiǎn)便。我喜歡Spark與其他大數(shù)據(jù)還有分析框架的原因在于：

l開(kāi)源與積極開(kāi)發(fā)

l不依賴(lài)工具，例如輸入數(shù)據(jù)與輸出數(shù)據(jù)不一定非要借助Hadoop

l獨(dú)立模式，啟動(dòng)迅速，易于部署

l大規(guī)模并行，易于添加節(jié)點(diǎn)

l支持多種輸入與輸出格式；比如可以讀取/寫(xiě)入MySQL（Java數(shù)據(jù)庫(kù)連接驅(qū)動(dòng)）與Parquet分列格式

但是，也有很多缺點(diǎn)：

l技術(shù)太新，會(huì)有一些bug和非法行為。很多錯(cuò)誤難以解釋。

l需要Java；Spark 1.5僅支持Java 7及以上版本。這也意味著需要額外內(nèi)存——合情合理。

l你需要通過(guò)“spark-submit”來(lái)運(yùn)行任務(wù)。

我認(rèn)為作為工具，Apache Spark十分不錯(cuò)，補(bǔ)足了MySQL在數(shù)據(jù)分析與商業(yè)智能方面的短板。

到此，相信大家對(duì)“如何使用Apache Spark和MySQL實(shí)現(xiàn)數(shù)據(jù)分析”有了更深的了解，不妨來(lái)實(shí)際操作一番吧！這里是億速云網(wǎng)站，更多相關(guān)內(nèi)容可以進(jìn)入相關(guān)頻道進(jìn)行查詢(xún)，關(guān)注我們，繼續(xù)學(xué)習(xí)！