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

官方文档

(0)

相关推荐