一.前言
前面我们学习的是使用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