top of page
Search
  • Learn with Shin

Spark - 窺探Big Data

Updated: Dec 7, 2021



最近這幾年Big Data(大數據)這個詞相當的流行,媒體新聞也會常常用到這個字眼。


當然,我們可以輕易的從網路上取得Big Data的定義,不過還是簡單的講一下,Big Data的Big(大)通常指的是資料的數量、速度以及類型(volume, velocity, variety),根據情況還有其他的特性。基本上當這些要素符合的時候,一般的單一電腦系統往往無法負荷。


因此,我們會利用所謂的分散式運算(distributed computing),簡單來說將資料拆成很多塊,並分給很多電腦來同時處理:一種團結力量大的概念。這些連在一起的大大小小的電腦形成一個所謂的Cluster。


實際上,一般來說自己會碰到Big Data的機會可能不多,大概也不會需要用到Cluster。但是這裡仍然希望給大家作一個參考,如果將來在工作上有遇到這個需求,或許可以提供一個方向😀。


今天要跟大家一起看的是在這種大數據概念下產生的一個工具(或是架構),叫做Spark(全名Apache Spark),或是更精準一點(因為我們用的是Python)-- 我們要來介紹PySpark(Python + Spark)。



我們這次主要想跟大家一起看看pyspark的一些syntax以及它的使用方式(相信這比談論一些理論來的容易懂)。話說,如果想試試看Cluster的感覺,可以參考像是Databricks,它提供的雲端平台讓我們可以快速的建立一個Cluster(記得使用Community Edition的免費版本喔)並在上面做執行Spark的程式。


PySpark


PySpark是Python用來跟Spark溝通的工具。首先,我們會需要pyspark這個library(注意:要在自己的local電腦上跑Spark的話,還需要安裝Spark本身跟做一些調整,完整安裝可以參考這裡)。


pip install pyspark

接下來的例子,我們會在Jupyter Notebook上操作。


首先第一個步驟,是建立SparkSession的物件。下圖可以看到,在SparkSession產生的同時你會看到一個叫做SparkContext的東西。這個SparkContext是Spark主要的核心,負責連結Spark Cluster。

一旦SparkContext建立後,可以利用瀏覽器進到localhost:4040去看到Spark的使用者介面。會看到像這樣👇的頁面,裡面記錄著你的Spark App的使用狀況:



話說關於處理效能,.defaultParallelism告訴你現在是用幾個CPU core在同時處理你的Spark job。

Cool!現在我們的Spark App已經在Local跑起來了。

RDD


接下來,我們來看一下Spark的核心資料結構 -- RDD(全名為Resilient Distributed Dataset)。

透過將資料轉成RDD,Spark能夠將要處理的資料拆解成許多小塊(partitioning)並分發到cluster中的各個電腦(或稱之為node)去處理。


舉個例子,這裡用簡單的Python的list來當作我們資料來源:

  • my_big_list是一個python的list,含有0~999999,一百萬個值在裡面。

  • 我們利用.sparkContext.parallelize將list轉成RDD的物件。

現在我們有了RDD,重點是該如何處理它呢?跟普通的Python list有什麼不同呢?


一般在處理像list這樣的集合時,可能動不動就會想要用for loop是吧?


但是在RDD的情況下,則通常會利用下面介紹的作業功能(operations):


map


假設,今天我想要將剛剛定義的rdd裡面的每一個數字都給他加倍,我可以利用.map這個method,範例如下:

  • map需要的argument是一個function物件。

  • lambda可以讓我們快速做出一個不需要名字的function(anonymous function),當我們只需要簡單的計算時,lambda是一個常用的方式。

  • map會回傳一個新的RDD物件。這裡的mapped_rdd就是一個RDD的物件。

  • .take可以給你指定數量的資料(以list的方式)



filter


接著,假設我想要挑出mapped_rdd裡面可以被4整除的數字,可以怎麼做呢?


我們會利用.flter這個功能:

  • filter也是需要function的物件作為argument,但是請注意這個function所回傳的值需要是一個boolean(True or False)值。

  • filter也是回傳一個新的RDD物件。因此這裡的filtered_rdd還是一個RDD物件。



reduce


再接下來,如果我想要算出這個filter_rdd的數值總和,可以怎麼做呢?

我可以利用.reduce這個功能:

  • reduce也是需要function的物件,但是注意這個function必須要接受兩個argument,並回傳單一數值。

  • reduce最後只會回傳一個數值。


map、filter、reduce在各個程式語言其實都存在。習慣這些技巧在Spark的作業中是相當重要的💪。



讀取外部資料來源


剛剛我們利用原生的list來生成RDD,但是其實大部分的情況會讀取來自外部的資料。Spark當然也有提供方便的API讓我們可以做到這一點。


譬如說,當需要的讀取的資料來源是一個text檔案的時候,我們可以利用.sparkContext.TextFile這個功能,檔案範例如下:


README
.txt
Download TXT • 5KB

下面的例子我們同時用.filter來截取得文件中每一行中含有spark的句子。


當然除了lambda,我們也可以用一般def的方式來定義,下面的例子可以取得每一行的字數:


Spark DataFrame


接下來,我們來看一下Spark另一種資料類型,叫做DataFrame,這可能算是最常用的Spark資料類型。Spark DataFrame就像pandas的DataFrame一樣是table格式的概念,但是想像它可以拆解成許多小塊來被同時處理。


下面的例子,我們讀進一個csv的檔案並生成一個Spark DataFrame:



操作上,跟pandas的DataFrame略有不同,但是概念大同小異。


譬如說,如果你想要看某個Column的時候,會用.select(column_name)。要將DataFrame秀出來的話,我們會用.show這個功能。

然後如果你想要篩選data的時候,可以用.filter(此filter跟之前看的rdd.filter是不一樣的喔)這個method,使用方法如下:


再者,如果你對SQL比較熟悉的話,DataFrame還可以讓你利用SQL的Query寫法來選擇資料喔:


關於dataframe的更多API可以參考官方文件



小結


今天我們快速介紹了Spark以及pyspark,並帶大家看了一些簡單的例子。希望對Spark好奇的你有一點幫助。想要了解更多細節或是例子的話可以參考Spark的官方文件


實作上,我們會用到Spark是因為有需要處理龐大的資料量。因此如果說資料本身不大的話其實是不需要用到Spark,但是理解這個工具的存在目的可以讓我們有更多的選項。尤其如果你已經使用過pandas的話,相信應該在概念上及操作方式上有許多相似之處,也會降低學習門檻。


當然,今天看到的只是Spark的冰山一角。Spark提供了資料分析處理的諸多元素,包括SparkSQL(處理table格式資料)、Spark Streaming(處理即時資料)、MLlib(機器學習)、Graph X(圖形資料),來因應當今科技(state of the art)的各種需求。希望透過這篇文章給有興趣的朋友一點參考喔~😊

Post: Blog2_Post
bottom of page