PySpark与GraphFrames的安装与使用环境搭建过程

pyspark环境搭建

配置hadoop

spark访问本地文件并执行运算时，可能会遇到权限问题或是dll错误。这是因为spark需要使用到hadoop的winutils和hadoop.dll，首先我们必须配置好hadoop相关的环境。可以到github下载：https://github.com/4ttty/winutils

gitcode提供了镜像加速：https://gitcode.net/mirrors/4ttty/winutils

我选择了使用这个仓库提供的最高的hadoop版本3.0.0将其解压到d:\deploy\hadoop-3.0.0目录下，然后配置环境变量：

我们还需要将对应的hadoop.dll复制到系统中，用命令表达就是：

不过这步需要拥有管理员权限才可以操作。

为了能够在任何地方使用winutils命令工具，将%hadoop_home%\bin目录加入环境变量中：

安装pyspark与java

首先，我们安装spark当前（2022-2-17）的最新版本：

需要注意pyspark的版本决定了jdk的最高版本，例如假如安装2.4.5版本的pyspark就只能安装1.8版本的jdk，否则会报出java.lang.illegalargumentexception: unsupported class file major version 55的错误。

这是因为pyspark内置了scala，而scala是基于jvm的编程语言，scala与jdk的版本存在兼容性问题，jdk与scala的版本兼容性表：

jdk version	minimum scala versions	recommended scala versions
17	2.13.6, 2.12.15 (forthcoming)	2.13.6, 2.12.15 (forthcoming)
16	2.13.5, 2.12.14	2.13.6, 2.12.14
13, 14, 15	2.13.2, 2.12.11	2.13.6, 2.12.14
12	2.13.1, 2.12.9	2.13.6, 2.12.14
11	2.13.0, 2.12.4, 2.11.12	2.13.6, 2.12.14, 2.11.12
8	2.13.0, 2.12.0, 2.11.0, 2.10.2	2.13.6, 2.12.14, 2.11.12, 2.10.7
6, 7	2.11.0, 2.10.0	2.11.12, 2.10.7

当前3.2.1版本的pyspark内置的scala版本为2.12.15，意味着jdk17与其以下的所有版本都支持。

这里我依然选择安装jdk8的版本：

测试一下：

jdk11的详细安装教程（jdk1.8在官网只有安装包，无zip绿化压缩包）：

绿化版java11的环境配置与python调用java
https://xxmdmst.blog.****.net/article/details/118366166

graphframes安装

pip安装当前最新的graphframes：

然后在官网下载graphframes的jar包。

下载地址：https://spark-packages.org/package/graphframes/graphframes

由于安装的pyspark版本是3.2，所以这里我选择了这个jar包：

然后将该jar包放入pyspark安装目录的jars目录下：

pyspark安装位置可以通过pip查看：

使用方法

学习pyspark的最佳路径是官网：https://spark.apache.org/docs/latest/quick-start.html

在下面的网页，官方提供了在线jupyter：

https://spark.apache.org/docs/latest/api/python/getting_started/index.html

启动spark并读取数据

本地模式启动spark：

sparksession输出的内容中包含了spark的web页面，新标签页打开页面后大致效果如上。

点击environment选项卡可以查看当前环境中的变量：

启动hive支持

找到pyspark的安装位置，例如我的电脑在d:\miniconda3\lib\site-packages\pyspark

手动创建conf目录并将hive-site.xml配置文件复制到其中。如果hive使用了mysql作为原数据库，则还需要将mysql对应的驱动jar包放入spark的jars目录下。

创建spark会话对象时可通过enablehivesupport()开启hive支持：

spark访问hive自己创建的表有可能会出现如下的权限报错：

caused by: java.lang.runtimeexception: the root scratch dir: /tmp/hive on hdfs s
hould be writable. current permissions are: rwx------

是因为当前用户不具备对\tmp\hive的操作权限：

把\tmp\hive目录的权限改为777即可顺利访问：

spark的dataframe与rdd

从spark2.x开始将rdd和dataframe的api统一抽象成dataset，dataframe就是dataset[row]，rdd则是dataset.rdd。可以将dataframe理解为包含结构化信息的rdd。

将含row的rdd转换为dataframe只需要调用todf方法或sparksession的createdataframe方法即可，也可以传入schema覆盖类型或名称设置。

dataframe的基础api

dataframe默认支持dsl风格语法，例如：

将dataframe注册成表或视图之后即可进行纯sql操作：

pyspark可以直接很方便的注册udf并直接使用：

执行结果：

[row(length='4')]
[row(length='3')]

rdd的简介

dataframe的本质是对rdd的包装，可以理解为dataframe=rdd[row]+schema。

rdd（a resilient distributed dataset）叫做弹性可伸缩分布式数据集，是spark中最基本的数据抽象。它代表一个不可变、自动容错、可伸缩性、可分区、里面的元素可并行计算的集合。

在每一个rdd内部具有五大属性：

• 具有一系列的分区
• 一个计算函数操作于每一个切片
• 具有一个对其他rdd的依赖列表
• 对于 key-value rdds具有一个partitioner分区器
• 存储每一个切片最佳计算位置

一组分片（partition），即数据集的基本组成单位。对于rdd来说，每个分片都会被一个计算任务处理，并决定并行计算的粒度。用户可以在创建rdd时指定rdd的分片个数，如果没有指定，那么就会采用默认值。默认值就是程序所分配到的cpu core的数目。

**一个计算每个分区的函数。**spark中rdd的计算是以分片为单位的，每个rdd都会实现compute函数以达到这个目的。compute函数会对迭代器进行复合，不需要保存每次计算的结果。

**rdd之间的依赖关系。**rdd的每次转换都会生成一个新的rdd，所以rdd之间就会形成类似于流水线一样的前后依赖关系。在部分分区数据丢失时，spark可以通过这个依赖关系重新计算丢失的分区数据，而不是对rdd的所有分区进行重新计算。

**一个partitioner，即rdd的分片函数。**当前spark中实现了两种类型的分片函数，一个是基于哈希的hashpartitioner，另外一个是基于范围的rangepartitioner。只有对于于key-value的rdd，才会有partitioner，非key-value的rdd的parititioner的值是none。partitioner函数不但决定了rdd本身的分片数量，也决定了parent rdd shuffle输出时的分片数量。

**一个列表，存储存取每个partition的优先位置（preferred location）。**对于一个hdfs文件来说，这个列表保存的就是每个partition所在的块的位置。按照“移动数据不如移动计算”的理念，spark在进行任务调度的时候，会尽可能地将计算任务分配到其所要处理数据块的存储位置。

rdd的api概览

rdd包含transformation api和 action api，transformation api都是延迟加载的只是记住这些应用到基础数据集上的转换动作，只有当执行action api时这些转换才会真正运行。

transformation api产生的两类rdd最重要，分别是mappartitionsrdd和shuffledrdd。

产生mappartitionsrdd的算子有map、keyby、keys、values、flatmap、mapvalues 、flatmapvalues、mappartitions、mappartitionswithindex、glom、filter和filterbyrange 。其中用的最多的是map和flatmap，但任何产生mappartitionsrdd的算子都可以直接使用mappartitions或mappartitionswithindex实现。

产生shuffledrdd的算子有combinebykeywithclasstag、combinebykey、aggregatebykey、foldbykey 、reducebykey 、distinct、groupbykey、groupby、partitionby、sortbykey 和 repartitionandsortwithinpartitions。

combinebykey到groupbykey 底层均是调用combinebykeywithclasstag方法：

三个重要参数的含义：

• createcombiner：根据每个分区的第一个元素操作产生一个初始值
• mergevalue：对每个分区内部的元素进行迭代合并
• mergecombiners：对所有分区的合并结果进行合并

groupbykey的partitioner未指定时会传入默认的defaultpartitioner。例如：

aggregatebykey：每个分区使用zerovalue作为初始值，迭代每一个元素用seqop进行合并，对所有分区的结果用combop进行合并。例如：

reducebykey ：每个分区迭代每一个元素用func进行合并，对所有分区的结果用func再进行合并，例如：

action api有：

动作	含义
reduce(func)	通过func函数聚集rdd中的所有元素，这个功能必须是课交换且可并联的
collect()	在驱动程序中，以数组的形式返回数据集的所有元素
count()	返回rdd的元素个数
first()	返回rdd的第一个元素（类似于take(1)）
take(n)	返回一个由数据集的前n个元素组成的数组
takesample(withreplacement*,*num, [seed])	返回一个数组，该数组由从数据集中随机采样的num个元素组成，可以选择是否用随机数替换不足的部分，seed用于指定随机数生成器种子
takeordered(n, [ordering])	排序并取前n个元素
saveastextfile(path)	将数据集的元素以textfile的形式保存到hdfs文件系统或者其他支持的文件系统，对于每个元素，spark将会调用tostring方法，将它装换为文件中的文本
saveassequencefile(path)	将数据集中的元素以hadoop sequencefile的格式保存到指定的目录下，可以使hdfs或者其他hadoop支持的文件系统。
saveasobjectfile(path)	将rdd中的元素用nullwritable作为key,实际元素作为value保存为sequencefile格式
countbykey()	针对(k,v)类型的rdd，返回一个(k,int)的map，表示每一个key对应的元素个数。
foreach(func)	在数据集的每一个元素上，运行函数func进行更新。

spark模拟实现mapreduce版wordcount：

各类rdd

• shuffledrdd ：表示需要走shuffle过程的网络传输
• coalescedrdd ：用于将一台机器的多个分区合并成一个分区
• cartesianrdd ：对两个rdd的所有元素产生笛卡尔积
• mappartitionsrdd ：用于对每个分区的数据进行特定的处理
• cogroupedrdd ：用于将2~4个rdd，按照key进行连接聚合
• subtractedrdd ：用于对2个rdd求差集
• unionrdd和partitionerawareunionrdd ：用于对2个rdd求并集
• zippedpartitionsrdd2：zip拉链操作产生的rdd
• zippedwithindexrdd：给每一个元素标记一个自增编号
• partitionwisesampledrdd：用于对rdd的元素按照指定的百分比进行随机采样

当我们需要给datafream添加自增列时，可以使用zipwithuniqueid方法：

api用法详情可参考：https://spark.apache.org/docs/latest/api/python/reference/api/pyspark.rdd.html#pyspark.rdd

cache&checkpoint

rdd通过persist方法或cache方法可以将前面的计算结果缓存，但是并不是这两个方法被调用时立即缓存，而是触发后面的action时，该rdd将会被缓存在计算节点的内存中，并供后面重用。

checkpoint的源码注释可以看到：

• 标记该rdd作为检查点。
• 它将被保存在通过sparkcontext#setcheckpointdir方法设置的检查点目录中
• 它所引用的所有父rdd引用将全部被移除
• 这个方法在这个rdd上必须在所有job执行前运行。
• 强烈建议将这个rdd缓存在内存中，否则这个保存文件的计算任务将重新计算。

从中我们得知，在执行checkpoint方法时，最好同时，将该rdd缓存起来，否则，checkpoint也会产生一个计算任务。

graphframes 的用法

graphframe是将spark中的graph算法统一到dataframe接口的graph操作接口，为scala、java和python提供了统一的图处理api。

graphframes是开源项目，源码工程如下：https://github.com/graphframes/graphframes

可以参考：

• 官网：https://graphframes.github.io/graphframes/docs/_site/index.html
• graphframes用户指南-python — databricks文档：https://docs.databricks.com/spark/latest/graph-analysis/graphframes/user-guide-python.html

在graphframes中图的顶点(vertex)和边(edge)都是以dataframe形式存储的：

• 顶点dataframe：必须包含列名为“id”的列，用于作为顶点的唯一标识
• 边dataframe：必须包含列名为“src”和“dst”的列，根据唯一标识id标识关系

创建图的示例：

graphframe提供三种视图：

获取顶点的度、入度和出度：

motif finding (模式发现)

示例：

假设我们要想给用户推荐关注的人，可以找出这样的关系：a关注b，b关注c，但是a未关注c。找出这样的关系就可以把c推荐给a：

结果：

+---+---+
|  a|  c|
+---+---+
|  e|  c|
|  e|  a|
|  d|  b|
|  a|  d|
|  f|  b|
|  d|  e|
|  a|  f|
|  a|  c|
+---+---+

motif在查找路径过程的过程中，还可以沿着路径携带状态。例如我们想要找出关系链有4个顶点，而且其中3条边全部都是"friend"关系：

结果：

+---------------+--------------+---------------+--------------+---------------+--------------+---------------+
|              a|            ab|              b|            bc|              c|            cd|              d|
+---------------+--------------+---------------+--------------+---------------+--------------+---------------+
| {a, alice, 34}|{a, e, friend}|{e, esther, 32}|{e, d, friend}| {d, david, 29}|{d, a, friend}| {a, alice, 34}|
|{e, esther, 32}|{e, d, friend}| {d, david, 29}|{d, a, friend}| {a, alice, 34}|{a, b, friend}|   {b, bob, 36}|
| {d, david, 29}|{d, a, friend}| {a, alice, 34}|{a, e, friend}|{e, esther, 32}|{e, d, friend}| {d, david, 29}|
|{e, esther, 32}|{e, d, friend}| {d, david, 29}|{d, a, friend}| {a, alice, 34}|{a, e, friend}|{e, esther, 32}|
+---------------+--------------+---------------+--------------+---------------+--------------+---------------+

subgraphs 子图

可以直接过滤其顶点或边，dropisolatedvertices()方法用于删除孤立没有连接的点：

还可以基于模式发现获取到的边创建subgraphs ：

graphframes支持的graphx算法

• pagerank：查找图中的重要顶点。
• 广度优先搜索(bfs)：查找从一组顶点到另一组顶点的最短路径
• 连通组件(connectedcomponents)：为具备连接关系的顶点分配相同的组件id
• 强连通组件(stronglyconnectedconponents)：根据每个顶点的强连通分量分配scc。
• 最短路径(shortest paths)：查找从每个顶点到目标顶点集的最短路径。
• 三角形计数(trianglecount)：计算每个顶点所属的三角形的数量，经常用于确定组的稳定性（相互连接的数量代表了稳定性）或作为其他网络度量（如聚类系数）的一部分，在社交网络分析中用来检测社区。
• 标签传播算法（lpa）：检测图中的社区。

pagerank算法：

结果：

+---+-------+---+-------------------+
| id|   name|age|           pagerank|
+---+-------+---+-------------------+
|  b|    bob| 36| 2.7025217677349773|
|  c|charlie| 30| 2.6667877057849627|
|  a|  alice| 34| 0.4485115093698443|
|  e| esther| 32| 0.3613490987992571|
|  f|  fanny| 36|0.32504910549694244|
|  d|  david| 29|0.32504910549694244|
|  g|  gabby| 60|0.17073170731707318|
+---+-------+---+-------------------+

可以设置起始顶点：

广度优先搜索bfs：

搜索从姓名叫esther到年龄小于32的最小路径：

可以指定只能在指定的边搜索：

connected components 连通组件：

必须先设置检查点：

结果：

+---+-------+---+------------+
| id|   name|age|   component|
+---+-------+---+------------+
|  a|  alice| 34|412316860416|
|  b|    bob| 36|412316860416|
|  c|charlie| 30|412316860416|
|  d|  david| 29|412316860416|
|  e| esther| 32|412316860416|
|  f|  fanny| 36|412316860416|
|  g|  gabby| 60|146028888064|
+---+-------+---+------------+

可以看到仅g点在一个连通区域内，可以调用dropisolatedvertices()方法，删除这种孤立的没有连接的点：

结果：

+---+-------+---+------------+
| id|   name|age|   component|
+---+-------+---+------------+
|  a|  alice| 34|412316860416|
|  b|    bob| 36|412316860416|
|  c|charlie| 30|412316860416|
|  d|  david| 29|412316860416|
|  e| esther| 32|412316860416|
|  f|  fanny| 36|412316860416|
+---+-------+---+------------+

strongly connected components 强连通组件：

shortest paths 最短路径：

每个顶点到a或d的最短路径：

triangle count 三角形计数：

说明顶点a/e/d构成三角形。

标签传播算法(lpa)：

pyspark3.x与pandas融合

pyspark从3.0版本开始出现了pandas_udf装饰器、applyinpandas和mapinpandas，基于这些方法，我们就可以使用熟悉的pandas的语法处理spark对象的数据。

首先创建几条测试数据，并启动 apache arrow：

自定义udf和udaf

pyspark暂不支持自定义udtf。

使用pandas_udf装饰器我们可以创建出基于pandas的udf自定义函数，在dsl的语法中可以被直接使用：

注册函数和视图后，可以直接在sql中使用：

结果均为：

+-------+
|product|
+-------+
|    1.0|
|    2.0|
|    6.0|
|   10.0|
|   20.0|
+-------+

还支持聚合函数和窗口函数：

注册到udf之后同样可以直接使用sql实现：

结果均为：

+--------+
| mean_v |
+--------+
|     4.2|
+--------+

+---+--------+
| id| mean_v |
+---+--------+
|  1|     1.5|
|  2|     6.0|
+---+--------+

+---+----+------+
| id|   v|mean_v|
+---+----+------+
|  1| 1.0|   1.5|
|  1| 2.0|   1.5|
|  2| 3.0|   6.0|
|  2| 5.0|   6.0|
|  2|10.0|   6.0|
+---+----+------+

分组聚合与join

applyinpandas需要在datafream调用groupby之后才能使用：

结果：

+---+----+----+----+
| id|   v|  v1|  v2|
+---+----+----+----+
|  1| 1.0|-0.5| 2.5|
|  1| 2.0| 0.5| 3.5|
|  2| 3.0|-3.0| 9.0|
|  2| 5.0|-1.0|11.0|
|  2|10.0| 4.0|16.0|
+---+----+----+----+

subtract_mean函数接收的是对应id的dataframe数据，schema指定了返回值的名称和类型列表。

通过以下代码我们可以知道，applyinpandas可以借助cogroup进行表连接：

示例：

map迭代

执行以下代码：

后台看到执行结果为：

0 [[2.0, 5.0]]                                                     
0 [[2.0, 3.0]]
0 [[1.0, 1.0]]
0 [[1.0, 2.0]]
0 [[2.0, 10.0]]

前台结果几乎保持原样。可以知道iterator是一个分区迭代器，迭代出当前分区的每一行数据都被封装成一个pandas对象。

pyspark与pandas的交互

将spark的datafream对象转换为原生的pandas对象只需调用topandas()方法即可：

将原生的pandas对象转换为spark对象可以使用spark的*方法：

习惯使用pandas的童鞋，还可以直接使用pandas-on-spark，在spark3.2.0版本时已经匹配到pandas 1.3版本的api。通过pandas-on-spark，我们可以完全用pandas的api操作数据，而底层执行却是spark的并行化。

使用pandas-on-spark最好设置一下环境变量：

将spark对象转换为pandas-on-spark对象：

pandas-on-spark对象也可以还原成spark对象：

另外spark提供直接将文件读取成pandas-on-spark对象的api，例如：

ps对象与原生pandas对象的api几乎完全一致。

ps对象相对于原生pandas对象的api几乎一致，同时还支持一些强悍的功能，例如直接以sql形式访问：

{pdf}访问了变量名为pdf的pandas-on-spark对象。

到此这篇关于pyspark与graphframes的安装与使用的文章就介绍到这了,更多相关pyspark与graphframes使用内容请搜索以前的文章或继续浏览下面的相关文章希望大家以后多多支持！