软件

1、anaconda(python3.6)

2、spark-2.4.3-bin-hadoop2.7(spark版本要和集群上的一样)

3、JDK1.8

python环境配置

pip install pyspark

这里如果遇到安装超时的情况采用以下命令

pip --default-timeout=100 install pyspark

或

pip --default-timeout=100 install -U pyspark

超时或者发生no matching distribution found for XXX 错误可以试着更换国内镜像源

pip install pyspark --default-timeout=100 -i https://pypi.tuna.tsinghua.edu.cn/simple

国内一些镜像源：

阿里云 http://mirrors.aliyun.com/pypi/simple/

中国科技大学 https://pypi.mirrors.ustc.edu.cn/simple/

豆瓣(douban) http://pypi.douban.com/simple/

清华大学 https://pypi.tuna.tsinghua.edu.cn/simple/

中国科学技术大学 http://pypi.mirrors.ustc.edu.cn/simple/

没有py4j也得装下

pip install py4j

spark环境配置

1、下载。spark下载地址：http://spark.apache.org/downloads.html

2、解压。下载好后解压 ，注意安装路径不能有空格。

3、配置环境变量。添加环境变量SPARK_HOME=spark安装路径(比如我的是D:\Application\spark-2.4.3-bin-hadoop2.7)，在Path中添加%SPARK_HOME%\bin;

4、验证。cmd到spark的bin目录下输入命令：spark-submit.cmd --help，出现以下内容说明spark安装成功

注意：spark集群默认是安装python2环境，需要python3环境还得装 并且默认环境切到python3

java环境配置

1、下载并安装好jdk1.8

2、添加环境变量JAVA_HOME=D:\Application\Java\jdk1.8.0_172，在Path中添加%JAVA_HOME%\bin;%JAVA_HOME%\jre\bin;%JAVA_HOME%\lib;%JAVA_HOME%\lib\tools.jar;

pyspark连接本地集群代码测试

from pyspark import SparkConf

from pyspark.sql import SparkSession

import traceback

appname = "test"#任务名称

master ="local"#单机模式设置

'''

local: 所有计算都运行在一个线程当中，没有任何并行计算，通常我们在本机执行一些测试代码，或者练手，就用这种模式。

local[K]: 指定使用几个线程来运行计算，比如local[4]就是运行4个worker线程。通常我们的cpu有几个core，就指定几个线程，最大化利用cpu的计算能力

local[*]: 这种模式直接帮你按照cpu最多cores来设置线程数了。

'''

try:

conf = SparkConf().setAppName(appname).setMaster(master)#spark资源配置

spark=SparkSession.builder.config(conf=conf).getOrCreate()

sc=spark.sparkContext

words = sc.parallelize(

["scala",

"java",

"hadoop",

"spark",

"akka",

"spark vs hadoop",

"pyspark",

"pyspark and spark"

])

counts = words.count()

print("Number of elements in RDD is %i" % counts)

sc.stop()

print('计算成功！')

except:

sc.stop()

traceback.print_exc()#返回出错信息

print('连接出错！')

运行结果：

pyspark连接远程集群代码测试

from pyspark import SparkConf

from pyspark.sql import SparkSession

import traceback

import os

os.environ["PYSPARK_PYTHON"]="/usr/bin/python3"#集群上pyspark的python版本指向python3

appname = "test"#任务名称

master ="spark://XXX.XXX.XX.XX:XXXX" #"spark://host:port"

'''

standalone模式:spark://host:port,Spark会自己负责资源的管理调度

mesos模式:mesos://host:port

yarn模式:由于很多时候我们需要和mapreduce使用同一个集群，所以都采用Yarn来管理资源调度，这也是生产环境大多采用yarn模式的原因。yarn模式又分为yarn cluster模式和yarn client模式：

yarn cluster: 这个就是生产环境常用的模式，所有的资源调度和计算都在集群环境上运行。

yarn client: 这个是说Spark Driver和ApplicationMaster进程均在本机运行，而计算任务在cluster上。

'''

spark_driver_host="XXX.XXX.XX.XX"#本地主机ip

try:

conf = SparkConf().setAppName(appname).setMaster(master).set("spark.driver.host",spark_driver_host )

spark=SparkSession.builder.config(conf=conf).getOrCreate()

sc=spark.sparkContext

words = sc.parallelize(

["scala",

"java",

"hadoop",

"spark",

"akka",

"spark vs hadoop",

"pyspark",

"pyspark and spark"

])

counts = words.count()

print("Number of elements in RDD is %i" % counts)

sc.stop()

print('计算成功！')

except:

sc.stop()

traceback.print_exc()#返回出错信息

print('连接出错！')

结果一样

如果对你有帮助，请点下赞，予人玫瑰手有余香！