作者：Pinar Ersoy
翻譯：孫韬淳
校對：陳振東
本文約2500字，建議閱讀10分鐘
本文通過介紹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
編輯：于騰凱
校對：洪舒越
譯者簡介
孫韬淳，首都師範大學大四在讀，主修遙感科學與技術。目前專注于基本知識的掌握和提升，期望在未來有機會探索資料科學在地學應用的衆多可能性。愛好之一為翻譯創作，在業餘時間加入到THU資料派平台的翻譯志願者小組，希望能和大家一起交流分享，共同進步。
—完—
關注清華-青島資料科學研究院官方微信公衆平台“ THU資料派 ”及姊妹号“ 資料派THU ”擷取更多講座福利及優質内容。