搭建Spark计算平台+python操作Spark

一、Spark安装及服务启动

Apache Spark是一种快速的集群计算技术，专为快速计算而设计。它基于Hadoop MapReduce，它扩展了MapReduce模型，以有效地将其用于更多类型的计算，包括交互式查询和流处理。 Spark的主要特性是它的内存中集群计算，提高了应用程序的处理速度（Spark 因为 RDD 是基于内存的，可以比较容易切成较小的块来处理。如果能对这些小块处理得足够快，就能达到低延时的效果）。

比起 Hadoop MapReduce， Spark 本质上就是基于内存的更快的批处理，然后用足够快的批处理来实现各种场景

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))

官方文档