PySpark和SparkSQL基礎：如何利用Python編程執行Spark（附代碼）

本文通過介紹Apache Spark在Python中的應用來講解如何利用PySpark包執行常用函數來進行數據處理工作。

 

Apache Spark是一個對開發者提供完備的庫和API的集群計算系統，並且支持多種語言，包括Java，Python，R和Scala。SparkSQL相當於Apache Spark的一個模塊，在DataFrame API的幫助下可用來處理非結構化數據。

通過名為PySpark的Spark Python API，Python實現了處理結構化數據的Spark編程模型。

這篇文章的目標是展示如何通過PySpark運行Spark並執行常用函數。

Python編程語言要求一個安裝好的IDE。最簡單的方式是通過Anaconda使用Python，因其安裝了足夠的IDE包，並附帶了其他重要的包。

1、下載Anaconda並安裝PySpark

通過這個鏈接，你可以下載Anaconda。你可以在Windows，macOS和Linux操作系統以及64位/32位圖形安裝程序類型間選擇。我們推薦安裝Python的最新版本。

 

Anaconda的安裝頁面（https://www.anaconda.com/distribution/）

下載好合適的Anaconda版本后，點擊它來進行安裝，安裝步驟在Anaconda Documentation中有詳細的說明。

安裝完成時，Anaconda導航主頁（Navigator Homepage）會打開。因為只是使用Python，僅需點擊“Notebook”模塊中的“Launch”按鈕。

 

Anaconda導航主頁

為了能在Anaconda中使用Spark，請遵循以下軟件包安裝步驟。

第一步：從你的電腦打開“Anaconda Prompt”終端。

第二步：在Anaconda Prompt終端中輸入“conda install pyspark”並回車來安裝PySpark包。

第三步：在Anaconda Prompt終端中輸入“conda install pyarrow”並回車來安裝PyArrow包。

當PySpark和PyArrow包安裝完成后，僅需關閉終端，回到Jupyter Notebook，並在你代碼的最頂部導入要求的包。

import pandas as pd

from pyspark.sql import SparkSession

from pyspark.context import SparkContext

from pyspark.sql.functions

import *from pyspark.sql.types

import *from datetime import date, timedelta, datetime

import time

2、初始化SparkSession

首先需要初始化一個Spark會話（SparkSession）。通過SparkSession幫助可以創建DataFrame，並以表格的形式注冊。其次，可以執行SQL表格，緩存表格，可以閱讀parquet/json/csv/avro數據格式的文檔。

sc = SparkSession.builder.appName("PysparkExample")\    

.config ("spark.sql.shuffle.partitions", "50")\    

.config("spark.driver.maxResultSize","5g")\    

.config ("spark.sql.execution.arrow.enabled", "true")\    

.getOrCreate()

想了解SparkSession每個參數的詳細解釋，請訪問pyspark.sql.SparkSession。

3、創建數據框架

一個DataFrame可被認為是一個每列有標題的分布式列表集合，與關系數據庫的一個表格類似。在這篇文章中，處理數據集時我們將會使用在PySpark API中的DataFrame操作。

你可以從https://www.kaggle.com/cmenca/new-york-times-hardcover-fiction-best-sellers中下載Kaggle數據集。

3.1、從Spark數據源開始

DataFrame可以通過讀txt，csv，json和parquet文件格式來創建。在本文的例子中，我們將使用.json格式的文件，你也可以使用如下列舉的相關讀取函數來尋找並讀取text，csv，parquet文件格式。

#Creates a spark data frame called as raw_data.

#JSON

dataframe = sc.read.json('dataset/nyt2.json')

#TXT FILES#

dataframe_txt = sc.read.text('text_data.txt')

#CSV FILES#

dataframe_csv = sc.read.csv('csv_data.csv')

#PARQUET FILES#

dataframe_parquet = sc.read.load('parquet_data.parquet')

4、重復值

表格中的重復值可以使用dropDuplicates()函數來消除。

dataframe = sc.read.json('dataset/nyt2.json')

dataframe.show(10)

 

使用dropDuplicates()函數后，我們可觀察到重復值已從數據集中被移除。

dataframe_dropdup = dataframe.dropDuplicates() dataframe_dropdup.show(10)

 

5、查詢

查詢操作可被用於多種目的，比如用“select”選擇列中子集，用“when”添加條件，用“like”篩選列內容。接下來將舉例一些最常用的操作。完整的查詢操作列表請看Apache Spark文檔。

5.1、“Select”操作

可以通過屬性（“author”）或索引（dataframe[‘author’]）來獲取列。

#Show all entries in title column

dataframe.select("author").show(10)

#Show all entries in title, author, rank, price columns

dataframe.select("author", "title", "rank", "price").show(10)

 

第一個結果表格展示了“author”列的查詢結果，第二個結果表格展示多列查詢。

5.2、“When”操作

在第一個例子中，“title”列被選中並添加了一個“when”條件。

# Show title and assign 0 or 1 depending on title

dataframe.select("title",when(dataframe.title != 'ODD HOURS',

1).otherwise(0)).show(10)

 

展示特定條件下的10行數據

在第二個例子中，應用“isin”操作而不是“when”，它也可用於定義一些針對行的條件。

# Show rows with specified authors if in the given options

dataframe [dataframe.author.isin("John Sandford",

"Emily Giffin")].show(5)

 

5行特定條件下的結果集

5.3、“Like”操作

在“Like”函數括號中，%操作符用來篩選出所有含有單詞“THE”的標題。如果我們尋求的這個條件是精確匹配的，則不應使用%算符。

# Show author and title is TRUE if title has " THE " word in titles

dataframe.select("author", "title",

dataframe.title.like("% THE %")).show(15)

 

title列中含有單詞“THE”的判斷結果集

5.4、“startswith”-“endswith”

StartsWith指定從括號中特定的單詞/內容的位置開始掃描。類似的，EndsWith指定了到某處單詞/內容結束。兩個函數都是區分大小寫的。

dataframe.select("author", "title",

dataframe.title.startswith("THE")).show(5)

dataframe.select("author", "title",

dataframe.title.endswith("NT")).show(5)

 

對5行數據進行startsWith操作和endsWith操作的結果。

5.5、“substring”操作

Substring的功能是將具體索引中間的文本提取出來。在接下來的例子中，文本從索引號（1,3），（3,6）和（1,6）間被提取出來。

dataframe.select(dataframe.author.substr(1

, 3).alias("title")).show(5)

dataframe.select(dataframe.author.substr(3

, 6).alias("title")).show(5)

dataframe.select(dataframe.author.substr(1

, 6).alias("title")).show(5)

 

分別顯示子字符串為（1,3），（3,6），（1,6）的結果

6、增加，修改和刪除列

在DataFrame API中同樣有數據處理函數。接下來，你可以找到增加/修改/刪除列操作的例子。

6.1、增加列

# Lit() is required while we are creating columns with exact

values.

dataframe = dataframe.withColumn('new_column',

F.lit('This is a new column'))

display(dataframe)

 

在數據集結尾已添加新列

6.2、修改列

對於新版DataFrame API，withColumnRenamed()函數通過兩個參數使用。

# Update column 'amazon_product_url' with 'URL'

dataframe = dataframe.withColumnRenamed('amazon_product_url', 'URL')

dataframe.show(5)

 

“Amazon_Product_URL”列名修改為“URL”

6.3、刪除列

列的刪除可通過兩種方式實現：在drop()函數中添加一個組列名，或在drop函數中指出具體的列。兩個例子展示如下。

dataframe_remove = dataframe.drop("publisher",

"published_date").show(5)

dataframe_remove2=dataframe \

.drop(dataframe.publisher).drop(dataframe.published_date).show(5)

 

“publisher”和“published_date”列用兩種不同的方法移除。

7、數據審閱

存在幾種類型的函數來進行數據審閱。接下來，你可以找到一些常用函數。想了解更多則需訪問Apache Spark doc。

# Returns dataframe column names and data types

dataframe.dtypes

# Displays the content of dataframe

dataframe.show()

# Return first n rows

dataframe.head()

# Returns first row

dataframe.first()

# Return first n rows

dataframe.take(5)

# Computes summary statistics

dataframe.describe().show()

# Returns columns of dataframe

dataframe.columns

# Counts the number of rows in dataframe

dataframe.count()

# Counts the number of distinct rows in dataframe

dataframe.distinct().count()

# Prints plans including physical and logical

dataframe.explain(4)

8、“GroupBy”操作

通過GroupBy()函數，將數據列根據指定函數進行聚合。

# Group by author, count the books of the authors in the groups

dataframe.groupBy("author").count().show(10)

 

作者被以出版書籍的數量分組

9、“Filter”操作

通過使用filter()函數，在函數內添加條件參數應用篩選。這個函數區分大小寫。

# Filtering entries of title

# Only keeps records having value 'THE HOST'

dataframe.filter(dataframe["title"] == 'THE HOST').show(5)

 

標題列經篩選后僅存在有“THE HOST”的內容，並顯示5個結果。

10、缺失和替換值

對每個數據集，經常需要在數據預處理階段將已存在的值替換，丟棄不必要的列，並填充缺失值。pyspark.sql.DataFrameNaFunction庫幫助我們在這一方面處理數據。舉例如下。

# Replacing null values

dataframe.na.fill()

dataFrame.fillna()

dataFrameNaFunctions.fill()

# Returning new dataframe restricting rows with null valuesdataframe.na.drop()

dataFrame.dropna()

dataFrameNaFunctions.drop()

# Return new dataframe replacing one value with another

dataframe.na.replace(5, 15)

dataFrame.replace()

dataFrameNaFunctions.replace()

11、重分區

在RDD（彈性分布數據集）中增加或減少現有分區的級別是可行的。使用repartition(self,numPartitions)可以實現分區增加，這使得新的RDD獲得相同/更高的分區數。分區縮減可以用coalesce(self, numPartitions, shuffle=False)函數進行處理，這使得新的RDD有一個減少了的分區數（它是一個確定的值）。請訪問Apache Spark doc獲得更多信息。

# Dataframe with 10 partitions

dataframe.repartition(10).rdd.getNumPartitions()

# Dataframe with 1 partition

dataframe.coalesce(1).rdd.getNumPartitions()

12、嵌入式運行SQL查詢

原始SQL查詢也可通過在我們SparkSession中的“sql”操作來使用，這種SQL查詢的運行是嵌入式的，返回一個DataFrame格式的結果集。請訪問Apache Spark doc獲得更詳細的信息。

# Registering a table

dataframe.registerTempTable("df")

sc.sql("select * from df").show(3)

sc.sql("select \               

CASE WHEN description LIKE '%love%' THEN 'Love_Theme' \               

WHEN description LIKE '%hate%' THEN 'Hate_Theme' \               

WHEN description LIKE '%happy%' THEN 'Happiness_Theme' \               

WHEN description LIKE '%anger%' THEN 'Anger_Theme' \               

WHEN description LIKE '%horror%' THEN 'Horror_Theme' \               

WHEN description LIKE '%death%' THEN 'Criminal_Theme' \               

WHEN description LIKE '%detective%' THEN 'Mystery_Theme' \               

ELSE 'Other_Themes' \               END Themes \       

from df").groupBy('Themes').count().show()

13、輸出

13.1、數據結構

DataFrame API以RDD作為基礎，把SQL查詢語句轉換為低層的RDD函數。通過使用.rdd操作，一個數據框架可被轉換為RDD，也可以把Spark Dataframe轉換為RDD和Pandas格式的字符串同樣可行。

# Converting dataframe into an RDD

rdd_convert = dataframe.rdd

# Converting dataframe into a RDD of string

dataframe.toJSON().first()

# Obtaining contents of df as Pandas

dataFramedataframe.toPandas()

 

不同數據結構的結果

13.2、寫並保存在文件中

任何像數據框架一樣可以加載進入我們代碼的數據源類型都可以被輕易轉換和保存在其他類型文件中，包括.parquet和.json。請訪問Apache Spark doc尋求更多保存、加載、寫函數的細節。

# Write & Save File in .parquet format

dataframe.select("author", "title", "rank", "description") \

.write \

.save("Rankings_Descriptions.parquet")

 

當.write.save()函數被處理時，可看到Parquet文件已創建。

# Write & Save File in .json format

dataframe.select("author", "title") \

.write \

.save("Authors_Titles.json",format="json")

 

當.write.save()函數被處理時，可看到JSON文件已創建。

13.3、停止SparkSession

Spark會話可以通過運行stop()函數被停止，如下。

# End Spark Session

sc.stop()

代碼和Jupyter Notebook可以在我的GitHub上找到。

歡迎提問和評論！

參考文獻：

1. http://spark.apache.org/docs/latest/

2. https://docs.anaconda.com/anaconda/

原文標題：

PySpark and SparkSQL Basics

How to implement Spark with Python Programming

原文鏈接：

https://towardsdatascience.com/pyspark-and-sparksql-basics-6cb4bf967e53