PySpark Documentation — PySpark 3.3.0 documentationhttps://spark.apache.org/docs/latest/api/python/

---

?前言 - Spark

---

1. 運行pyspark的各種方式

1，通過pyspark進入pyspark單機交互式環境。
這種方式一般用來測試代碼。
也可以指定jupyter或者ipython為交互環境。

2，通過spark-submit提交Spark任務到集群運行。
這種方式可以提交Python腳本或者Jar包到集群上讓成百上千個機器運行任務。
這也是工業界生產中通常使用spark的方式。

3，通過zepplin notebook交互式執行。
zepplin是jupyter notebook的apache對應產品。

4, Python安裝findspark和pyspark庫。
可以在jupyter和其它Python環境中像調用普通庫一樣地調用pyspark庫。

2. Spark基本概念

• RDD：是彈性分布式數據集（Resilient Distributed Dataset）的簡稱，是分布式內存的一個抽象概念，提供了一種高度受限的共享內存模型。
• DAG：是Directed Acyclic Graph（有向無環圖）的簡稱，反映RDD之間的依賴關系。
• Driver Program：控制程序，負責為Application構建DAG圖。
• Cluster Manager：集群資源管理中心，負責分配計算資源。
• Worker Node：工作節點，負責完成具體計算。
• Executor：是運行在工作節點（Worker Node）上的一個進程，負責運行Task，并為應用程序存儲數據。
• Application：用戶編寫的Spark應用程序，一個Application包含多個Job。
• Job：作業，一個Job包含多個RDD及作用于相應RDD上的各種操作。
• Stage：階段，是作業的基本調度單位，一個作業會分為多組任務，每組任務被稱為“階段”。
• Task：任務，運行在Executor上的工作單元，是Executor中的一個線程。
• 總結：Application由多個Job組成，Job由多個Stage組成，Stage由多個Task組成。Stage是作業調度的基本單位。

3. PySpark架構設計

pyspark,為了不破壞Spark已有的運行時架構，Spark在外圍包裝一層Python API。在Driver端，借助Py4j實現Python和Java的交互，進而實現通過Python編寫Spark應用程序。在Executor端，則不需要借助Py4j，因為Executor端運行的Task邏輯是由Driver發過來的，那是序列化后的字節碼。

4.?Spark運行流程

Application首先被Driver構建DAG圖并分解成Stage。

然后Driver向Cluster Manager申請資源。

Cluster Manager向某些Work Node發送征召信號。

被征召的Work Node啟動Executor進程響應征召，并向Driver申請任務。

Driver分配Task給Work Node。

Executor以Stage為單位執行Task，期間Driver進行監控。

Driver收到Executor任務完成的信號后向Cluster Manager發送注銷信號。

Cluster Manager向Work Node發送釋放資源信號。

Work Node對應Executor停止運行。

5.?RDD數據結構

RDD，Resilient Distributed Dataset，彈性分布式數據集，它是記錄的只讀分區集合，是Spark的基本數據結構。RDD代表一個不可變、可分區、里面的元素可并行計算的集合。
一般有兩種方式創建RDD，第一種是讀取文件中的數據生成RDD，第二種則是通過將內存中的對象并行化得到RDD。

#通過讀取文件生成RDD rdd = sc.textFile("hdfs://data_warehouse/test/data") #通過將內存中的對象并行化得到RDD arr = [1,2,3,4,5] rdd = sc.parallelize(arr)

RDD的操作有兩種類型，即Transformation操作和Action操作。轉換操作是從已經存在的RDD創建一個新的RDD，而行動操作是在RDD上進行計算后返回結果到 Driver。
Transformation操作都具有 Lazy 特性，即 Spark 不會立刻進行實際的計算，只會記錄執行的軌跡，只有觸發Action操作的時候，它才會根據 DAG 圖真正執行。

操作確定了RDD之間的依賴關系。
RDD之間的依賴關系有兩種類型，即窄依賴和寬依賴。窄依賴時，父RDD的分區和子RDD的分區的關系是一對一或者多對一的關系。而寬依賴時，父RDD的分區和子RDD的分區是一對多或者多對多的關系。
寬依賴關系相關的操作一般具有shuffle過程，即通過一個Patitioner函數將父RDD中每個分區上key不同的記錄分發到不同的子RDD分區。

依賴關系確定了DAG切分成Stage的方式。
切割規則：從后往前，遇到寬依賴就切割Stage。

RDD之間的依賴關系形成一個DAG有向無環圖，DAG會提交給DAGScheduler，DAGScheduler會把DAG劃分成相互依賴的多個stage，劃分stage的依據就是RDD之間的寬窄依賴。遇到寬依賴就劃分stage,每個stage包含一個或多個task任務。然后將這些task以taskSet的形式提交給TaskScheduler運行。
?

6. 代碼例子

import pyspark from pyspark import SparkContext, SparkConfconf = SparkConf().setAppName("test").setMaster("local[4]") sc = SparkContext(conf=conf) print("spark version:",pyspark.__version__) rdd = sc.parallelize(["hello","spark"]) print(rdd.reduce(lambda x,y:x+' '+y))#wordcount例子 rdd_line = sc.textFile("/home/data/hello.txt") rdd_word = rdd_line.flatMap(lambda x:x.split(" ")) rdd_one = rdd_word.map(lambda t:(t,1)) rdd_count = rdd_one.reduceByKey(lambda x,y:x+y) rdd_count.collect()

一、RDD編程

• 創建RDD
• 常用Action操作
• 常用Transformation操作
• 常用PairRDD的轉換操作
• 緩存操作
• 共享變量
• 分區操作

二、SparkSQL編程

• RDD和DataFrame的對比
• 創建DataFrame
• DataFrame保存成文件
• DataFrame的API交互
• DataFrame的SQL交互

三、Spark性能調優方法

一般來說，如果有可能，用戶應當盡可能多地使用SparkSQL以取得更好的性能。

主要原因是SparkSQL是一種聲明式編程風格，背后的計算引擎會自動做大量的性能優化工作。基于RDD的Spark的性能調優屬于坑非常深的領域，并且很容易踩到。

可以用下面三個公式來近似估計spark任務的執行時間：

可以用下面二個公式來說明spark在executor上的內存分配：

四、分布式實現代碼?

五、MLlib機器學習

• DataFrame: MLlib中數據的存儲形式，其列可以存儲特征向量，標簽，以及原始的文本，圖像。
• Transformer：轉換器。具有transform方法。通過附加一個或多個列將一個DataFrame轉換成另外一個DataFrame。
• Estimator：估計器。具有fit方法。它接受一個DataFrame數據作為輸入后經過訓練，產生一個轉換器Transformer。
• Pipeline：流水線。具有setStages方法。順序將多個Transformer和1個Estimator串聯起來，得到一個流水線模型。

from pyspark.ml.feature import Tokenizer,HashingTF from pyspark.ml.classification import LogisticRegression from pyspark.ml.evaluation import MulticlassClassificationEvaluator,BinaryClassificationEvaluator from pyspark.ml import Pipeline,PipelineModel from pyspark.ml.linalg import Vector from pyspark.sql import Row

Spark MLlib的分布式訓練方法有下面幾個弊端：

采用全局廣播的方式，在每輪迭代前廣播全部模型參數。眾所周知Spark的廣播過程非常消耗帶寬資源，特別是當模型的參數規模過大時，廣播過程和在每個節點都維護一個權重參數副本的過程都是極消耗資源的過程，這導致了Spark在面對復雜模型時的表現不佳；

采用阻斷式的梯度下降方式，每輪梯度下降由最慢的節點決定。從上面的分析可知，Spark MLlib的mini batch的過程是在所有節點計算完各自的梯度之后，逐層Aggregate最終匯總生成全局的梯度。也就是說，如果由于數據傾斜等問題導致某個節點計算梯度的時間過長，那么這一過程將block其他所有節點無法執行新的任務。這種同步阻斷的分布式梯度計算方式，是Spark MLlib并行訓練效率較低的主要原因；

Spark MLlib并不支持復雜網絡結構和大量可調超參。事實上，Spark MLlib在其標準庫里只支持標準的多層感知機神經網絡的訓練，并不支持RNN，LSTM等復雜網絡結構，而且也無法選擇不同的activation function等大量超參。這就導致Spark MLlib在支持深度學習方面的能力欠佳。

1. 數據操作

Sql.dataframe?

spark = SparkSession.builder \.appName("spark") \.getOrCreate()df = spark.read.csv('fraudTrain.csv',header=True) df.printSchema()

數據查看

df.count() #查看計數 df.show(truncate=False) #dataframe顯示

數據處理

df.take(1) df.select('col')

2. 特征工程

缺失值

df.na.drop() # df.dropna() df.na.fill()

編碼

from pyspark.ml.feature import Binarizer#二值化 binarizer = Binarizer(threshold=10, inputCol='humidty', outputCol= 'label') df = binarizer.transform(df) df.select('humidty','label').show(4)

標準化/歸一化

#特征 from pyspark.ml.feature import VectorAssembler assemble = VectorAssembler(inputCols=featuralist, outputCol='features') df = assemble.transform(df)#標準化 from pyspark.ml.feature import StandardScaler scale = StandardScaler(inputCol='features', outputCol='standardized') scaler = scale.fit(df) df = scaler.transform(df)

3. 特征選擇

Chi-Square selector

4. 機器學習模型

from pyspark.ml import Pipelinefrom pyspark.ml.clustering import KMeans from pyspark.ml.classification import LogisticRegressiondt from pyspark.ml.regression import DecisionTreeRegressor#邏輯回歸 blor = LogisticRegression() blorModel = blor.fit(df) blorModel.evaluate(df).accuracy == blorModel.summary.accuracyblorModel.predict(df) blorModel.transform(df).prediction#決策樹回歸 dt = DecisionTreeRegressor(maxDepth=2) model = dt.fit(df) model.featureImportances

5. 模型評估

from pyspark.ml.evaluation import BinaryClassificationEvaluatorevaluator = BinaryClassificationEvaluator(rawPredictionCol= 'rawPrediction') evaluator.evaluate(dataset)

6. 文本挖掘

詞頻統計?TF-IDF

TF是詞頻(Term Frequency)，IDF是逆文本頻率指數(Inverse Document Frequency)

TF-IDF的主要思想是：如果某個詞或短語在一篇文章中出現的頻率TF高，并且在其他文章中很少出現，則認為此詞或者短語具有很好的類別區分能力，適合用來分類。

TF-IDF實際上是：TF * IDF。某一特定文件內的高詞語頻率，以及該詞語在整個文件集合中的低文件頻率，可以產生出高權重的TF-IDF。因此，TF-IDF傾向于過濾掉常見的詞語，保留重要的詞語。

from pyspark.ml.feature import HashingTF, IDF, Tokenizer, CountVectorizer from pyspark.ml.feature import StringIndexer#1. HashingTF + IDF + Logistic Regression. #2. CountVectorizer + IDF + Logistic Regression.#分詞 tokenizer = Tokenizer(inputCol='text',outputCol="words") df= tokenizer.transform(df)# numFeatures特征數上限，不同word的數量 hashtf = HashingTF(numFeatures=2**16, inputCol='words', outputCol="tf") hashingTF.transform(df).head().features # => SparseVector(10, {5: 1.0, 7: 1.0, 8: 1.0})# 逆向文件頻率 idf= IDF(minDocFreq=5, inputCol='tf', outputCol="features") model = idf.fit(df) model.transform(df)# 將string列映射為 label [0, numLabels) 0是頻率最高的 # 也可以將numeric列映射為帶標簽string indexer=StringIndexer(inputCol="target", outputCol="label",stringOrderType="frequencyDesc") model = indexer.fit(df) df= model.transform(df)# CountVectorizer()

六、基于pySpark的Pandas API

References

PySpark | Spark框架簡述 | Spark環境搭建_跟烏龜賽跑的博客-CSDN博客_pyspark

【Pyspark教程】SQL、MLlib、Core等模塊基礎使用_山頂夕景的博客-CSDN博客_pyspark

【PySpark調優】 3萬字長文，PySpark入門級學習教程，框架思維_CSDN博客

總結

以上是生活随笔為你收集整理的PySpark入门的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。