PySpark安装使用

概览

这里不直接安装原生的Spark，因为目前没必要，所以仅仅记录下PySpark的安装和使用过程中遇到的坑。

安装

在Spark的官方的安装界面就提到了PySpark的安装方式就是直接运行pip install pyspark。这当然是可以的，但是我们一般不直接这样做，主要是因为Anaconda的依赖问题。在默认的Base环境中混用pip install和conda install很容易把环境搞坏掉，使得后面根本没法升级。

所以养成好的习惯，一般不要去动Base, 需要配置为xx而用的Python环境就搞一个虚拟环境出来，这样就很容易管理。

这里我们就用conda建立一个Python3.7的名字为pyspark的环境conda create --name pyspark python=3.7.之后激活此环境conda activate pyspark。此时直接在终端输入pyspark即可。

注意，我在运行的时候遇到一个小问题，就是此时打开的之前安装的Spark版本2.3.0，而不是最新的2.4.4.查了下主要问题在一个环境变量SPARK_HOME，我之前指定到了那个2.3.0的版本(通过env | grep spark发现)。我们在/etc/profile将其注释掉之后重启电脑即可。此时激活虚拟环境后，运行pyspark会直接进入pip安装的新的版本。

默认启动IPython

上面运行pyspark会直接运行python，而我想让他进入IPython，显然后者更为方便一些。修改步骤如下，首先which pyspark定位spark的位置，发现在/home/shensir/anaconda3/envs/pyspark/bin/pyspark之后直接执行gedit /home/shensir/anaconda3/envs/pyspark/bin/pyspark对其进行修改：

1 2	export PYSPARK_DRIVER_PYTHON export PYSPARK_DRIVER_PYTHON_OPTS

定位到上面两行，为第一行的环境变量指定为ipython即可，如下：

1 2	export PYSPARK_DRIVER_PYTHON='ipython' export PYSPARK_DRIVER_PYTHON_OPTS

之后再次运行pyspark就会直接进入IPython.

连接MySQL数据库

在看一个教程的时候，尝试了从MySQL数据库创建RDD，但是报错[java.lang.ClassNotFoundException: com.mysql.jdbc.Driver](https://www.cnblogs.com/spicy/p/9754123.html),后参考[Cant connect to Mysql database from pyspark, getting jdbc error发现是少了驱动文件，看了下数据库是8.0.17版本，所以在MySQL Connector/J (Archived Versions)选择8.0.17和Platform Independent下载mysql-connector-java-8.0.17.tar.gz文件，之后解压，找到mysql-connector-java-8.0.17.jar文件，将其复制到/home/shensir/anaconda3/envs/pyspark/lib/python3.7/site-packages/pyspark/jars文件夹下，重启终端继续即可。

下面贴一下PySpark连接MySQL数据的代码：

from pyspark.sql import SparkSession

spark = SparkSession \
	.builder \
    	.appName('Python Spark creates RDD') \
        .config('Spark.some.config.option', 'some-value') \
        .getOrCreate()

df = spark.read.format("jdbc").options(
	url="jdbc:mysql://localhost:3306/数据库名",
	driver="com.mysql.cj.jdbc.Driver",
	dbtable = "表名",
	user="用户名",
	password="密码").load()

print(df.show())

从HDFS读取文件

首先要保证服务开启start-dfs.sh，可以jps查看各个节点是否启动成功。同时使用hdfs dfs -ls -R查看已经加入HDFS的文件。我们也可以用命令hdfs dfs -moveFromLocal data/WineData.csv /user/shensir来将本地文件添加到HDFS，之后再次通过hdfs dfs -ls -R查看是否添加成功。

注，开启服务后我们也可以在浏览器查看HDFS文件目录，我这里是在http://localhost:9870/explorer.html#/user/shensir查看，但是注意在代码中读取的位置依旧是用9000端口。

之后可以在PySpark中读取:

from pyspark.context import SparkContext
from pyspark.sql import HiveContext

sc = SparkContext('local','example')
hc = HiveContext(sc)
data = sc.textFile("hdfs://localhost:9000/user/shensir/WineData.csv")
print(data.first())

# RDD->DataFrame
header = data.first()
df = data \
    .filter(lambda row: row != header) \
    .map(lambda row: [float(ch) for ch in row.split(',')]) \
    .toDF(header.split(','))
print(df.show(5))