一、Spark安装及服务启动
Apache Spark是一种快速的集群计算技术，专为快速计算而设计。它基于Hadoop MapReduce，它扩展了MapReduce模型，以有效地将其用于更多类型的计算，包括交互式查询和流处理。 Spark的主要特性是它的内存中集群计算，提高了应用程序的处理速度（Spark 因为 RDD 是基于内存的，可以比较容易切成较小的块来处理。如果能对这些小块处理得足够快，就能达到低延时的效果）。
1、安装Scala
下载并解压Scala
cd /opt/scala
wget https://downloads.lightbend.com/scala/2.11.7/scala-2.11.7.tgz
tar -zxf scala-2.11.7.tgz
将Scala添加到环境变量
vi /etc/profile
在最后面添加
export SCALA_HOME=/opt/scala/scala-2.11.7
export PATH=$PATH:$SCALA_HOME/bin
激活配置
source /etc/profile
2、Spark下载
从官网下载和自己hadoop版本相匹配的spark安装包
3、解压安装文件并配置环境变量
（1）解压安装文件
解压安装文件到指定的的文件夹 /opt/spark
tar -zxvf spark-2.2.0-bin-hadoop2.7.tgz -C opt/spark
修改文件夹名字
cd /opt/spark/
mv spark-2.2.0-bin-hadoop2.7 spark-2.2.0
（2）配置环境变量
export SPARK_HOME=/opt/spark/spark-2.2.0
export PATH=$SPARK_HOME/bin:$PATH
4、配置Spark
需要修改的配置文件有两个
spark-env.sh，spark-defaults.conf
（1）配置spark-env.sh
cd /opt/spark/spark-2.2.0/conf
mv spark-env.sh.template spark-env.sh
vim spark-env.sh
配置如下内容
# 配置JAVA_HOME
export JAVA_HOME=/opt/java/jdk1.8.0_144
# 配置SCALA_HOME
export SCALA_HOME=/opt/scala/scala-2.11.7
# 配置HADOOP
export HADOOP_HOME=/opt/hadoop/hadoop-2.7.6/
export HADOOP_CONF_DIR=/opt/hadoop/hadoop-2.7.6/etc/hadoop
#定义管理端口
export SPARK_MASTER_WEBUI_PORT=8088
#定义master域名和端口
export SPARK_MASTER_HOST=spark-master
export SPARK_MASTER_PORT=7077 # 提交Application的端口
export SPARK_MASTER_IP=10.141.211.80
# 定义work节点的管理端口
export SPARK_WORKER_WEBUI_PORT=8088
# 每一个Worker最多可以使用的cpu core的个数，真实服务器如果有32个，可以设置为32个
export SPARK_WORKER_CORES=10
# 每一个Worker最多可以使用的内存，真实服务器如果有128G，你可以设置为100G
export SPARK_WORKER_MEMORY=4g
（2）配置spark-defaults.conf
cd /opt/spark/spark-2.2.0/conf
vim spark-defaults.conf
配置如下内容
spark.eventLog.enabled=true
spark.eventLog.compress=true
# 保存在本地
# spark.eventLog.dir=file://opt/hadoop/hadoop-2.7.6/logs/userlogs
# spark.history.fs.logDirectory=file://opt/hadoop/hadoop-2.7.6/logs/userlogs
# 保存在hdfs上
spark.eventLog.dir=hdfs://spark-master:9000/tmp/logs/root/logs
spark.history.fs.logDirectory=hdfs://spark-master:9000/tmp/logs/root/logs
spark.yarn.historyServer.address=spark-master:18080
5、启动Spark
sbin/start-all.sh
二、PySpark安装
pyspark是用来对接 spark的 Python 库
pip install pyspark
三、使用pyspark
1、SparkContext声明
SparkContext是任何spark功能的入口点。
from pyspark import SparkContext
sc = SparkContext("local", "First App")
2、一些基本操作
（1）count
返回RDD中的元素个数
from pyspark import SparkContext
sc = SparkContext("local", "count app")
words = sc.parallelize(
["scala",
"java",
"hadoop",
"spark",
"akka",
"spark vs hadoop",
"pyspark",
"pyspark and spark"
])
counts = words.count()
print("Number of elements in RDD -> %i" % counts)
（2）collect
返回RDD中的所有元素
from pyspark import SparkContext
sc = SparkContext("local", "collect app")
words = sc.parallelize(
["scala",
"java",
"hadoop",
"spark",
"akka",
"spark vs hadoop",
"pyspark",
"pyspark and spark"
])
coll = words.collect()
print("Elements in RDD -> %s" % coll)
（3）foreach
仅返回满足foreach内函数条件的元素。在下面的示例中，我们在foreach中调用print函数，该函数打印RDD中的所有元素。
from pyspark import SparkContext
sc = SparkContext("local", "ForEach app")
words = sc.parallelize (
["scala",
"java",
"hadoop",
"spark",
"akka",
"spark vs hadoop",
"pyspark",
"pyspark and spark"]
)
def f(x): print(x)
fore = words.foreach(f)
（4）filter
返回一个包含元素的新RDD，它满足过滤器内部的功能。在下面的示例中，我们过滤掉包含''spark'的字符串。
from pyspark import SparkContext
sc = SparkContext("local", "Filter app")
words = sc.parallelize(
["scala",
"java",
"hadoop",
"spark",
"akka",
"spark vs hadoop",
"pyspark",
"pyspark and spark"]
)
words_filter = words.filter(lambda x: 'spark' in x)
filtered = words_filter.collect()
print("Fitered RDD -> %s" % (filtered))
（5）map
通过将该函数应用于RDD中的每个元素来返回新的RDD。在下面的示例中，我们形成一个键值对，并将每个字符串映射为值1
from pyspark import SparkContext
sc = SparkContext("local", "Map app")
words = sc.parallelize(
["scala",
"java",
"hadoop",
"spark",
"akka",
"spark vs hadoop",
"pyspark",
"pyspark and spark"]
)
words_map = words.map(lambda x: (x, 1))
mapping = words_map.collect()
print("Key value pair -> %s" % (mapping))
（6）reduce
执行指定的可交换和关联二元操作后，将返回RDD中的元素。在下面的示例中，我们从运算符导入add包并将其应用于'num'以执行简单的加法运算。说白了和Python的reduce一样：假如有一组整数[x1,x2,x3]，利用reduce执行加法操作add，对第一个元素执行add后，结果为sum=x1,然后再将sum和x2执行add，sum=x1+x2，最后再将x2和sum执行add，此时sum=x1+x2+x3。
from pyspark import SparkContext
from operator import add
sc = SparkContext("local", "Reduce app")
nums = sc.parallelize([1, 2, 3, 4, 5])
adding = nums.reduce(add)
print("Adding all the elements -> %i" % (adding))
（7）join
它返回RDD，其中包含一对带有匹配键的元素以及该特定键的所有值。
from pyspark import SparkContext
sc = SparkContext("local", "Join app")
x = sc.parallelize([("spark", 1), ("hadoop", 4)])
y = sc.parallelize([("spark", 2), ("hadoop", 5)])
joined = x.join(y)
final = joined.collect()
print( "Join RDD -> %s" % (final))
