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

一、Spark安装及服务启动

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

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

1、安装Scala

下载并解压Scala

1
2
3
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添加到环境变量

1
vi /etc/profile

在最后面添加

1
2
export SCALA_HOME=/opt/scala/scala-2.11.7
export PATH=$PATH:$SCALA_HOME/bin

激活配置

1
source /etc/profile

2、Spark下载

官网下载和自己hadoop版本相匹配的spark安装包

3、解压安装文件并配置环境变量

(1)解压安装文件
解压安装文件到指定的的文件夹 /opt/spark

1
tar -zxvf spark-2.2.0-bin-hadoop2.7.tgz -C  opt/spark

修改文件夹名字

1
2
cd /opt/spark/
mv spark-2.2.0-bin-hadoop2.7 spark-2.2.0

(2)配置环境变量

1
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

1
2
3
cd /opt/spark/spark-2.2.0/conf
mv spark-env.sh.template spark-env.sh
vim spark-env.sh

配置如下内容

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
# 配置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

1
2
cd /opt/spark/spark-2.2.0/conf
vim spark-defaults.conf

配置如下内容

1
2
3
4
5
6
7
8
9
10
11
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

1
sbin/start-all.sh

二、PySpark安装

pyspark是用来对接 spark的 Python 库

1
pip install pyspark

三、使用pyspark

1、SparkContext声明

SparkContext是任何spark功能的入口点。

1
2
from pyspark import SparkContext
sc = SparkContext("local", "First App")

2、一些基本操作

(1)count
返回RDD中的元素个数

1
2
3
4
5
6
7
8
9
10
11
12
13
14
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中的所有元素

1
2
3
4
5
6
7
8
9
10
11
12
13
14
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中的所有元素。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
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’的字符串。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
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

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
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。

1
2
3
4
5
6
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,其中包含一对带有匹配键的元素以及该特定键的所有值。

1
2
3
4
5
6
7
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))

官方文档

0%