一.前言
前面我們學習的是使用Scala和Java開發Spark。最近補充了下Python基礎，那么就用Python開發下Spark。Python開發Spark簡稱PySpark。
二.環境准備
1.安裝Python環境
安裝方式有兩種
使用原生方式安裝
直接去官網下載，window下載 xxx_executable installer這樣的。
下載完后傻瓜式安裝（注意安裝的時候勾選將Python配置到環境。注意安裝路徑不要有中文，不要有空格。）
使用anaconda安裝
下載anaconda，然后傻瓜式安裝，安裝完后就會默認給安裝一個Python版本。
如果我們需要手動配置版本，請使用上一篇博文介紹，將Python下載后，手動配置下環境即可。
2.配置Spark Home
下載Spark並解壓，然后在系統變量里面添加變量名 SPARK_HOME ，變量值就是Spark解壓路徑，比如我 的F:\BigDataPack\spark\spark-2.4.5-bin-hadoop2.7 。
添加完后，再在Path里面添加 %SPARK_HOME%\bin 。然后確定即可。
3.安裝pyspark
pyspark是操作spark的庫。安裝方式有兩種：
使用pip安裝
直接在cmd運行，pip install pyspark
直接拷貝
進入 spark根目錄/python/lib 下面，將pyspark解壓，將解壓后的文件復制到
anaconda或者你手動安裝的python 的 Lib/site-packages 下面。
4. 安裝py4j
py4j是負責python和java通信的。安裝方式有兩種：
使用pip安裝
直接在cmd運行，pip install py4j
直接拷貝
進入 spark根目錄/python/lib 下面，將py4j解壓，將解壓后的文件復制到
anaconda或者你手動安裝的python 的 Lib/site-packages 下面。
三.使用案例
項目結構:
BrowserCard文件內容:
192.156.345.6 www.baidu.com
192.156.345.6 www.shuai7boy.vip
192.156.345.6 www.google.com
145.345.67.78 www.baidu.com
192.156.345.6 www.baidu.com
145.345.67.78 www.cnblogs.com
135.674.33.23 www.csdn.cn
words文件內容:
hello world
hello YiMing
hello world
案例1.計算wordcount
代碼如下：
from pyspark import SparkContext, SparkConf
if __name__ == '__main__':
conf = SparkConf().setAppName("word").setMaster("local")
sc = SparkContext(conf=conf)
rdd = sc.textFile("words")
result = rdd.flatMap(lambda line: line.split(" ")).map(lambda line: (line, 1)).reduceByKey(lambda a, b: a + b)
result.foreach(print)
運行結果：
E:\SoftCollection\Anaconda\Install\python.exe F:/Code/Python/PySparkTest/UV.py
Using Spark's default log4j profile: org/apache/spark/log4j-defaults.properties
Setting default log level to "WARN".
To adjust logging level use sc.setLogLevel(newLevel). For SparkR, use setLogLevel(newLevel).
[Stage 2:> (0 + 1) / 1]('www.baidu.com', 2)
('www.cnblogs.com', 1)
('www.csdn.cn', 1)
('www.google.com', 1)
('www.shuai7boy.vip', 1)
Process finished with exit code 0
案例2.計算pv
代碼如下:
from pyspark import SparkConf, SparkContext
if __name__ == '__main__':
conf = SparkConf().setMaster("local").setAppName("PV")
sc = SparkContext(conf=conf)
cards = sc.textFile("BrowserCard")
result = cards.map(lambda line: (line.split(" ")[1], 1)).reduceByKey(lambda a, b: a + b)
result.foreach(print)
運行結果：
E:\SoftCollection\Anaconda\Install\python.exe F:/Code/Python/PySparkTest/PV.py
Using Spark's default log4j profile: org/apache/spark/log4j-defaults.properties
Setting default log level to "WARN".
To adjust logging level use sc.setLogLevel(newLevel). For SparkR, use setLogLevel(newLevel).
[Stage 0:> (0 + 1) / 1]('www.baidu.com', 3)
('www.shuai7boy.vip', 1)
('www.google.com', 1)
('www.cnblogs.com', 1)
('www.csdn.cn', 1)
Process finished with exit code 0
案例3.計算uv
代碼如下:
from pyspark import SparkConf, SparkContext
if __name__ == '__main__':
conf = SparkConf().setMaster("local").setAppName("PV")
sc = SparkContext(conf=conf)
cards = sc.textFile("BrowserCard")
result = cards.map(lambda line: line.split(" ")[0] + "_" + line.split(" ")[1]).distinct().map(
lambda line: (line.split(
"_")[1], 1)).reduceByKey(lambda a, b: a + b).sortByKey();
result.foreach(print)
運行結果：
E:\SoftCollection\Anaconda\Install\python.exe F:/Code/Python/PySparkTest/UV.py
Using Spark's default log4j profile: org/apache/spark/log4j-defaults.properties
Setting default log level to "WARN".
To adjust logging level use sc.setLogLevel(newLevel). For SparkR, use setLogLevel(newLevel).
[Stage 0:> (0 + 1) / 1]('www.baidu.com', 2)
('www.cnblogs.com', 1)
('www.csdn.cn', 1)
('www.google.com', 1)
('www.shuai7boy.vip', 1)
Process finished with exit code 0