PySpark求解连通图问题

前文回顾&＃xff1a;

PySpark与GraphFrames的安装与使用
https://xxmdmst.blog.csdn.net/article/details/123009617

networkx快速解决连通图问题
https://xxmdmst.blog.csdn.net/article/details/123012333

前面我讲解了PySpark图计算库的使用以及纯python解决连通图问题的两个示例。这篇文章我们继续对上次的连通图问题改用PySpark实现。

需求1&＃xff1a;找社区

刘备和关羽有关系&＃xff0c;说明他们是一个社区&＃xff0c;刘备和张飞也有关系&＃xff0c;那么刘备、关羽、张飞归为一个社区&＃xff0c;以此类推。

对于这个连通图问题使用Pyspark如何解决呢&＃xff1f;

首先&＃xff0c;我们创建spark对象&＃xff1a;

from pyspark.sql import SparkSession, Row
from graphframes import GraphFramespark &＃61; SparkSession \.builder \.appName("PySpark") \.master("local[*]") \.getOrCreate()
sc &＃61; spark.sparkContext
# 设置检查点目录
sc.setCheckpointDir("checkpoint")


然后构建数据&＃xff1a;

data &＃61; [[&＃39;刘备&＃39;, &＃39;关羽&＃39;],[&＃39;刘备&＃39;, &＃39;张飞&＃39;],[&＃39;张飞&＃39;, &＃39;诸葛亮&＃39;],[&＃39;曹操&＃39;, &＃39;司马懿&＃39;],[&＃39;司马懿&＃39;, &＃39;张辽&＃39;],[&＃39;曹操&＃39;, &＃39;曹丕&＃39;]
]
data &＃61; spark.createDataFrame(data, ["人员", "相关人员"])
data.show()


&＃43;------&＃43;--------&＃43;
| 人员|相关人员|
&＃43;------&＃43;--------&＃43;
| 刘备| 关羽|
| 刘备| 张飞|
| 张飞| 诸葛亮|
| 曹操| 司马懿|
|司马懿| 张辽|
| 曹操| 曹丕|
&＃43;------&＃43;--------&＃43;


很明显原始数据就是图计算所要求的边数据&＃xff0c;只修改一下列名即可&＃xff1a;

edges &＃61; data.toDF("src", "dst")
edges.printSchema()


root|-- src: string (nullable &＃61; true)|-- dst: string (nullable &＃61; true)


下面我们开始构建顶点数据&＃xff1a;

vertices &＃61; (edges.rdd.flatMap(lambda x: x).distinct().map(lambda x: Row(x)).toDF(["id"])
)
vertices.show()


&＃43;------&＃43;
| id|
&＃43;------&＃43;
|诸葛亮|
| 刘备|
| 曹操|
|司马懿|
| 曹丕|
| 关羽|
| 张飞|
| 张辽|
&＃43;------&＃43;


下面使用spark的图计算 计算连通图&＃xff1a;

g &＃61; GraphFrame(vertices, edges)
result &＃61; g.connectedComponents().orderBy("component")
result.show()


&＃43;------&＃43;------------&＃43;
| id| component|
&＃43;------&＃43;------------&＃43;
|司马懿| 0|
| 张辽| 0|
| 曹丕| 0|
| 曹操| 0|
| 关羽|635655159808|
| 刘备|635655159808|
| 张飞|635655159808|
|诸葛亮|635655159808|
&＃43;------&＃43;------------&＃43;


可以看到结果中已经顺利将一个社区的成员通过一个相同的component标识出来&＃xff0c;成功解决需求。

需求2&＃xff1a;统一用户识别

abcde这5个字段表示mac地址&＃xff0c;ip地址&＃xff0c;device_id&＃xff0c;imei等唯一标识&＃xff0c;tags表示用户的标签。由于某些原因&＃xff0c;同一用户的唯一标识字段总是有几个字段存在缺失&＃xff0c;现在要求将同一个用户的数据都能识别出来&＃xff0c;同时将每个用户的标签进行合并。原始数据和结果模型示例如下&＃xff1a;

首先&＃xff0c;我们构建数据&＃xff1a;

df &＃61; spark.createDataFrame([[&＃39;a1&＃39;, None, &＃39;c1&＃39;, None, None, &＃39;tag1&＃39;],[None, None, &＃39;c1&＃39;, &＃39;d1&＃39;, None, &＃39;tag2&＃39;],[None, &＃39;b1&＃39;, None, &＃39;d1&＃39;, None, &＃39;tag3&＃39;],[None, &＃39;b1&＃39;, &＃39;c1&＃39;, &＃39;d1&＃39;, &＃39;e1&＃39;, &＃39;tag4&＃39;],[&＃39;a2&＃39;, &＃39;b2&＃39;, None, None, None, &＃39;tag1&＃39;],[None, &＃39;b4&＃39;, &＃39;c4&＃39;, None, &＃39;e4&＃39;, &＃39;tag1&＃39;],[&＃39;a2&＃39;, None, None, &＃39;d2&＃39;, None, &＃39;tag2&＃39;],[None, None, &＃39;c2&＃39;, &＃39;d2&＃39;, None, &＃39;tag3&＃39;],[None, &＃39;b3&＃39;, None, None, &＃39;e3&＃39;, &＃39;tag1&＃39;],[None, None, &＃39;c3&＃39;, None, &＃39;e3&＃39;, &＃39;tag2&＃39;],
], list("abcde")&＃43;["tags"])
df.show()


结果&＃xff1a;

&＃43;----&＃43;----&＃43;----&＃43;----&＃43;----&＃43;----&＃43;
| a| b| c| d| e|tags|
&＃43;----&＃43;----&＃43;----&＃43;----&＃43;----&＃43;----&＃43;
| a1|null| c1|null|null|tag1|
|null|null| c1| d1|null|tag2|
|null| b1|null| d1|null|tag3|
|null| b1| c1| d1| e1|tag4|
| a2| b2|null|null|null|tag1|
|null| b4| c4|null| e4|tag1|
| a2|null|null| d2|null|tag2|
|null|null| c2| d2|null|tag3|
|null| b3|null|null| e3|tag1|
|null|null| c3|null| e3|tag2|
&＃43;----&＃43;----&＃43;----&＃43;----&＃43;----&＃43;----&＃43;


接下来的思路依然跟上次一样&＃xff0c;首先为每一行数据分配一个唯一id&＃xff0c;然后对每个唯一标识的列&＃xff0c;根据是否一样构建行与行之间的连接关系&＃xff0c;所有的唯一标识列产生的连接关系共同作为图计算的边。

下面使用RDD的zipWithUniqueId方法为每一行产生一个唯一ID&＃xff0c;并将这个ID移动到最前&＃xff08;由于这个数据后面可能会多次被频繁使用所以缓存起来&＃xff09;&＃xff1a;

tmp &＃61; df.rdd.zipWithUniqueId().map(lambda x: (x[1], x[0]))
tmp.cache()
tmp.first()


(0, Row(a&＃61;&＃39;a1&＃39;, b&＃61;None, c&＃61;&＃39;c1&＃39;, d&＃61;None, e&＃61;None, tags&＃61;&＃39;tag1&＃39;))


根据唯一id构建顶点数据&＃xff1a;

vertices &＃61; tmp.map(lambda x: Row(x[0])).toDF(["id"])
vertices.show()


&＃43;---&＃43;
| id|
&＃43;---&＃43;
| 0|
| 1|
| 7|
| 2|
| 8|
| 3|
| 4|
| 10|
| 5|
| 11|
&＃43;---&＃43;


接下来&＃xff0c;构建边数据&＃xff1a;

def func(p):for k, ids in p:ids &＃61; list(ids)n &＃61; len(ids)if n <&＃61; 1:continuefor i in range(n-1):for j in range(i&＃43;1, n):yield (ids[i], ids[j])edges &＃61; []
keylist &＃61; list("abcde")
for key in keylist:data &＃61; tmp.mapPartitions(lambda area: [(row[key], i) for i, row in area if row[key]])edgeRDD &＃61; data.groupByKey().mapPartitions(func)edges.append(edgeRDD)
edgesDF &＃61; sc.union(edges).toDF(["src", "dst"])
edgesDF.cache()
edgesDF.show()


&＃43;---&＃43;---&＃43;
|src|dst|
&＃43;---&＃43;---&＃43;
| 8| 4|
| 7| 2|
| 0| 1|
| 0| 2|
| 1| 2|
| 4| 10|
| 1| 7|
| 1| 2|
| 7| 2|
| 5| 11|
&＃43;---&＃43;---&＃43;


可以看到所有的行号关系已经被成功获取。

下面使用图计算 计算出属于同一用户的行&＃xff1a;

gdf &＃61; GraphFrame(vertices, edgesDF)
components &＃61; gdf.connectedComponents()
components.show()


&＃43;---&＃43;---------&＃43;
| id|component|
&＃43;---&＃43;---------&＃43;
| 0| 0|
| 1| 0|
| 7| 0|
| 2| 0|
| 8| 4|
| 3| 3|
| 4| 4|
| 10| 4|
| 5| 5|
| 11| 5|
&＃43;---&＃43;---------&＃43;


有了行号和所归属的组唯一标识&＃xff0c;我们可以通过表连接获取原始数据的每一行所归属的component&＃xff1a;

result &＃61; tmp.cogroup(components.rdd) \.map(lambda pair: pair[1][0].data[0] &＃43; Row(pair[1][1].data[0])) \.toDF(df.schema.names&＃43;["component"])
result.cache()
result.show()


&＃43;----&＃43;----&＃43;----&＃43;----&＃43;----&＃43;----&＃43;---------&＃43;
| a| b| c| d| e|tags|component|
&＃43;----&＃43;----&＃43;----&＃43;----&＃43;----&＃43;----&＃43;---------&＃43;
| a1|null| c1|null|null|tag1| 0|
|null|null| c1| d1|null|tag2| 0|
|null| b1| c1| d1| e1|tag4| 0|
|null| b4| c4|null| e4|tag1| 3|
| a2|null|null| d2|null|tag2| 4|
|null| b3|null|null| e3|tag1| 5|
|null| b1|null| d1|null|tag3| 0|
| a2| b2|null|null|null|tag1| 4|
|null|null| c2| d2|null|tag3| 4|
|null|null| c3|null| e3|tag2| 5|
&＃43;----&＃43;----&＃43;----&＃43;----&＃43;----&＃43;----&＃43;---------&＃43;


可以看到我们已经成功的进行同一用户识别了&＃xff0c;剩下的只需要分组并使用pandas的逻辑合并数据&＃xff1a;

def func(pdf):row &＃61; pdf[keylist].bfill().head(1)row["tags"] &＃61; pdf.tags.str.cat(sep&＃61;",")return rowresult.groupBy("component").applyInPandas(func, schema&＃61;"a string, b string, c string, d string, e string, tags string"
).show()


&＃43;----&＃43;---&＃43;---&＃43;----&＃43;----&＃43;-------------------&＃43;
| a| b| c| d| e| tags|
&＃43;----&＃43;---&＃43;---&＃43;----&＃43;----&＃43;-------------------&＃43;
| a1| b1| c1| d1| e1|tag1,tag2,tag4,tag3|
|null| b4| c4|null| e4| tag1|
| a2| b2| c2| d2|null| tag2,tag1,tag3|
|null| b3| c3|null| e3| tag1,tag2|
&＃43;----&＃43;---&＃43;---&＃43;----&＃43;----&＃43;-------------------&＃43;


可以看到已经顺利得到需要的结果。

注意&＃xff1a;applyInPandas要求返回的结果必须是pandas的datafream对象&＃xff0c;所以相对之前的逻辑由.iloc[0]改成了.head(1)

如果你的spark不是3.X版本&＃xff0c;没有applyInPandas方法&＃xff0c;用原生rdd的方法则会麻烦很多&＃xff1a;

def func(pair):component, rows &＃61; pairkeyList &＃61; list("abcde")ids &＃61; {}for row in rows:for key in keylist:v &＃61; getattr(row, key)if v:ids[key] &＃61; vids.setdefault("tags", []).append(row.tags)result &＃61; []for key in keylist:result.append(ids.get(key))result.append(",".join(ids["tags"]))return resultresult2 &＃61; result.rdd.groupBy(lambda row: row.component).map(func).toDF(df.schema)
result2.cache()
result2.show()


结果也一样&＃xff1a;

&＃43;----&＃43;---&＃43;---&＃43;----&＃43;----&＃43;-------------------&＃43;
| a| b| c| d| e| tags|
&＃43;----&＃43;---&＃43;---&＃43;----&＃43;----&＃43;-------------------&＃43;
| a1| b1| c1| d1| e1|tag1,tag2,tag4,tag3|
|null| b4| c4|null| e4| tag1|
| a2| b2| c2| d2|null| tag2,tag1,tag3|
|null| b3| c3|null| e3| tag1,tag2|
&＃43;----&＃43;---&＃43;---&＃43;----&＃43;----&＃43;-------------------&＃43;