阿里云服务器centos7.2下安装Spark 2.2.1

 

安装Spark时版本选择与hadoop版本有关。

点击打开链接 进入后，选择一个能用的镜像。

 

一般情况下，这两个镜像都能用，

http://www-eu.apache.org/dist/spark/spark-2.2.1/spark-2.2.1-bin-hadoop2.7.tgz 

http://www-us.apache.org/dist/spark/spark-2.2.1/spark-2.2.1-bin-hadoop2.7.tgz 

1、下载

执行命令:

wget http://www-eu.apache.org/dist/spark/spark-2.2.1/spark-2.2.1-bin-hadoop2.7.tgz 

2、解压并更改文件名

tar zxvf spark-2.2.1-bin-hadoop2.7.tgz
mv spark-2.2.1-bin-hadoop2.7 spark  #方便后续操作，可以自行更改

3、配置环境变量：

vim /etc/profile

export SPARK_HOME=/usr/local/spark
export PATH=$PATH:$SPARK_HOME/bin

让修改后的配置文件生效：

source /etc/profile

4、运行spark

执行命令：

./bin/pyspark

开始编写Python程序。

在上面的操作过程中，会发现有很多INFO日志弹出。通过修spark文件中

cd /usr/local/spark/conf

cp log4j.properties.template log4j.properties
vim log4j.properties

把原来文档中的INFO改为WARN

log4j.rootCategory=WARN, console

然后按Esc，：wq保存退出。

可以通过修改配置文件改变pyspark默认选择的Python版本：

cd /usr/local/spark/conf
cp spark-env.sh.template spark-env.sh
vim spark-env.sh

添加如下命令到spark-env.sh文件中（放在第二行就可以）

export PYSPARK_PYTHON=/usr/local/bin/python3
export PYSPARK_DRIVER_PYTHON=/usr/local/bin/python3

保存退出。

重新运行pyspark。发现Python版本变为python3.6了。

如果想要使用IPython或者jupyter，在spark目录下执行以下命令：

PYSPARK_DRIVER_PYTHON=ipython ./bin/pyspark

或者

PYSPARK_DRIVER_PYTHON=jupyter
PYSPARK_DRIVER_PYTHON_OPTS=notebook ./bin/pyspark

5、在spark上运行python代码

5.1 Spark的主要抽象是一个名为Dataset的分布式项目集合。可以从Hadoop InputFormats（例如HDFS文件）或通过转换其他数据集来创建数据集。由于Python的动态特性，我们不需要在Python中强类型数据集。因此，Python中的所有数据集都是Dataset [Row]，我们称之为DataFrame与Pandas和R中的数据框概念一致。让我们从Spark源目录中的README文件的文本中创建一个新的DataFrame：

执行命令：

textFile = spark.read.text("README.md")

报错如下：

这是因为在使用相对路径时，系统默认是从hdfs://localhost:9000/usr/root目录下读取README.md文件的，但是README.md文件并不在这一目录下，所以spark.read.text()必须使用绝对路径。

修改命令如下：

textFile = spark.read.text("file:///usr/local/spark/README.md")

 

5.2可以通过调用某些操作直接从DataFrame获取值，也可以转换DataFrame以获取新值。

第一个命令输出的是DataFrame的行数，第二个命令输出的是DataFrame的第一行的内容。

 

5.3现在让我们将这个DataFrame转换为一个新的DataFrame。我们调用filter返回一个新的DataFrame，其中包含文件中的一行子集。

这个命令输出的是README.md文件中包含”Spark”该字符串的行数

 

5.4

这首先将一行映射为整数值，并将其别名为“numWords”，从而创建一个新的DataFrame。agg在该DataFrame上调用以查找最大的字数。参数select和agg都是Column，我们可以用来df.colName从DataFrame中获取一个列。我们还可以导入pyspark.sql.functions，它提供了很多方便的功能来从旧的列构建一个新的列。

 

 

参考

官方文档：http://spark.apache.org/docs/latest/quick-start.html

中文文档：http://spark.apachecn.org/docs/cn/2.2.0/quick-start.html

Spark 笔记|教程 —— Python API：http://www.myoak.info/post/10/