脚本之家 服务器常用软件
快捷导航
您的位置：首页 → 脚本专栏 → python → PySpark环境配置

windowns使用PySpark环境配置和基本操作

 更新时间：2021年05月17日 12:05:04   作者：Nick_Spider  
pyspark是Spark对Python的api接口,可以在Python环境中通过调用pyspark模块来操作spark，这篇文章主要介绍了windowns使用PySpark环境配置和基本操作，感兴趣的可以了解一下

下载依赖

首先需要下载hadoop和spark，解压，然后设置环境变量。
hadoop清华源下载
spark清华源下载

HADOOP_HOME => /path/hadoop
SPARK_HOME => /path/spark

安装pyspark。

pip install pyspark

基本使用

可以在shell终端，输入pyspark，有如下回显：

输入以下指令进行测试，并创建SparkContext，SparkContext是任何spark功能的入口点。

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

如果以上不会报错，恭喜可以开始使用pyspark编写代码了。
不过，我这里使用IDE来编写代码，首先我们先在终端执行以下代码关闭SparkContext。

>>> sc.stop()

下面使用pycharm编写代码，如果修改了环境变量需要先重启pycharm。
在pycharm运行如下程序，程序会起本地模式的spark计算引擎，通过spark统计abc.txt文件中a和b出现行的数量，文件路径需要自己指定。

from pyspark import SparkContext

sc = SparkContext("local", "First App")
logFile = "abc.txt"
logData = sc.textFile(logFile).cache()
numAs = logData.filter(lambda s: 'a' in s).count()
numBs = logData.filter(lambda s: 'b' in s).count()
print("Line with a:%i,line with b:%i" % (numAs, numBs))

运行结果如下：

20/03/11 16:15:57 WARN NativeCodeLoader: Unable to load native-hadoop library for your platform... using builtin-java classes where applicable
Using Spark's default log4j profile: org/apache/spark/log4j-defaults.properties
Setting default log level to "WARN".
To adjust logging level use sc.setLogLevel(newLevel). For SparkR, use setLogLevel(newLevel).
20/03/11 16:15:58 WARN Utils: Service 'SparkUI' could not bind on port 4040. Attempting port 4041.
Line with a:3,line with b:1

这里说一下，同样的工作使用python可以做，spark也可以做，使用spark主要是为了高效的进行分布式计算。
戳pyspark教程
戳spark教程

RDD

RDD代表Resilient Distributed Dataset，它们是在多个节点上运行和操作以在集群上进行并行处理的元素，RDD是spark计算的操作对象。
一般，我们先使用数据创建RDD，然后对RDD进行操作。
对RDD操作有两种方法：
Transformation（转换） - 这些操作应用于RDD以创建新的RDD。例如filter，groupBy和map。
Action（操作） - 这些是应用于RDD的操作，它指示Spark执行计算并将结果发送回驱动程序，例如count，collect等。

创建RDD

parallelize是从列表创建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"
     ])
print(words)

结果中我们得到一个对象，就是我们列表数据的RDD对象，spark之后可以对他进行操作。

ParallelCollectionRDD[0] at parallelize at PythonRDD.scala:195

Count

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"
     ])
print(words)

counts = words.count()
print("Number of elements in RDD -> %i" % counts)

返回结果：

Number of elements in RDD -> 8

Collect

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)

返回结果：

Elements in RDD -> ['scala', 'java', 'hadoop', 'spark', 'akka', 'spark vs hadoop', 'pyspark', 'pyspark and spark']

foreach

每个元素会使用foreach内的函数进行处理，但是不会返回任何对象。
下面的程序中，我们定义的一个累加器accumulator，用于储存在foreach执行过程中的值。

from pyspark import SparkContext
sc = SparkContext("local", "ForEach app")

accum = sc.accumulator(0)
data = [1, 2, 3, 4, 5]
rdd = sc.parallelize(data)


def increment_counter(x):
    print(x)
    accum.add(x)
 return 0

s = rdd.foreach(increment_counter)
print(s)  # None
print("Counter value: ", accum)

返回结果：

None
Counter value:  15

filter

返回一个包含元素的新RDD，满足过滤器的条件。

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

 

Fitered RDD -> ['spark', 'spark vs hadoop', 'pyspark', 'pyspark and 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"]
)


def g(x):
    for i in x:
        if "spark" in x:
            return i

words_filter = words.filter(g)
filtered = words_filter.collect()
print("Fitered RDD -> %s" % (filtered))

map

将函数应用于RDD中的每个元素并返回新的RDD。

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, "_{}".format(x)))
mapping = words_map.collect()
print("Key value pair -> %s" % (mapping))

返回结果：

Key value pair -> [('scala', 1, '_scala'), ('java', 1, '_java'), ('hadoop', 1, '_hadoop'), ('spark', 1, '_spark'), ('akka', 1, '_akka'), ('spark vs hadoop', 1, '_spark vs hadoop'), ('pyspark', 1, '_pyspark'), ('pyspark and spark', 1, '_pyspark and spark')]

Reduce

执行指定的可交换和关联二元操作后，然后返回RDD中的元素。

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

 这里的add是python内置的函数，可以使用ide查看：

def add(a, b):
    "Same as a + b."
    return a + b

reduce会依次对元素相加，相加后的结果加上其他元素，最后返回结果（RDD中的元素）。

Adding all the elements -> 15

Join

返回RDD，包含两者同时匹配的键，键包含对应的所有元素。

from pyspark import SparkContext


sc = SparkContext("local", "Join app")
x = sc.parallelize([("spark", 1), ("hadoop", 4), ("python", 4)])
y = sc.parallelize([("spark", 2), ("hadoop", 5)])
print("x =>", x.collect())
print("y =>", y.collect())
joined = x.join(y)
final = joined.collect()
print( "Join RDD -> %s" % (final))

返回结果：

x => [('spark', 1), ('hadoop', 4), ('python', 4)]
y => [('spark', 2), ('hadoop', 5)]
Join RDD -> [('hadoop', (4, 5)), ('spark', (1, 2))]

到此这篇关于windowns使用PySpark环境配置和基本操作的文章就介绍到这了,更多相关PySpark环境配置 内容请搜索脚本之家以前的文章或继续浏览下面的相关文章希望大家以后多多支持脚本之家！

您可能感兴趣的文章:

相关文章

  • YOLOv5改进教程之添加注意力机制

    YOLOv5改进教程之添加注意力机制

    注意力机制最先被用在NLP领域,Attention就是为了让模型认识到数据中哪一部分是最重要的,为它分配更大的权重,获得更多的注意力在一些特征上,让模型表现更好,这篇文章主要给大家介绍了关于YOLOv5改进教程之添加注意力机制的相关资料,需要的朋友可以参考下
    2022-06-06
  • 深入理解Python中的Contextlib库

    深入理解Python中的Contextlib库

    Python提供了一些内建的库以支持各种常见的编程任务，Contextlib库是其中之一，它提供了一些用于支持上下文管理协议（即with语句）的函数，这篇文章将详细介绍如何使用Contextlib库中的功能,需要的朋友可以参考下
    2023-06-06
  • python iloc和loc切片的实现

    python iloc和loc切片的实现

    本文主要介绍了python iloc和loc切片的实现，文中通过示例代码介绍的非常详细，对大家的学习或者工作具有一定的参考学习价值，需要的朋友们下面随着小编来一起学习学习吧
    2022-05-05
  • python 对给定可迭代集合统计出现频率,并排序的方法

    python 对给定可迭代集合统计出现频率,并排序的方法

    今天小编就为大家分享一篇python 对给定可迭代集合统计出现频率,并排序的方法，具有很好的参考价值，希望对大家有所帮助。一起跟随小编过来看看吧
    2018-10-10
  • 详解TensorFlow训练网络两种方式

    详解TensorFlow训练网络两种方式

    本文主要介绍了TensorFlow训练网络两种方式，一种是基于tensor(array)，另外一种是迭代器，文中通过示例代码介绍的非常详细，具有一定的参考价值，感兴趣的小伙伴们可以参考一下
    2021-12-12
  • Python编程中用close()方法关闭文件的教程

    Python编程中用close()方法关闭文件的教程

    这篇文章主要介绍了Python编程中用close()方法关闭文件的教程,是Python编程入门中的基础知识,需要的朋友可以参考下
    2015-05-05
  • Python退出While循环的3种方法举例详解

    Python退出While循环的3种方法举例详解

    在每次循环结束后,我们需要检查循环条件是否满足。如果条件满足,则继续执行循环体内的代码,否则退出循环,这篇文章主要给大家介绍了关于Python退出While循环的3种方法,需要的朋友可以参考下
    2023-10-10
  • Python3 能振兴 Python的原因分析

    Python3 能振兴 Python的原因分析

    我从Stephen A. Goss那读到关于了《Python 3正在毁灭Python》。这篇文章有不少精彩的论点，但我却并不认为Python 3是在毁灭Python，也不认为整个局面对Python一点也不利
    2014-11-11
  • paramiko模块安装和使用(远程登录服务器)

    paramiko模块安装和使用(远程登录服务器)

    paramiko是用python语言写的一个模块，遵循SSH2协议，支持以加密和认证的方式，进行远程服务器的连接,下面学习一下它的使用方法
    2014-01-01
  • Python编程快速上手——Excel表格创建乘法表案例分析

    Python编程快速上手——Excel表格创建乘法表案例分析

    这篇文章主要介绍了Python Excel表格创建乘法表,结合具体实例形式分析了Python接受cmd命令操作Excel文件创建乘法表相关实现技巧,需要的朋友可以参考下
    2020-02-02

最新评论