PySpark：如何将边列表DataFrame加载到GraphFrame与NetworkX图中

问题1：GraphFrame初始化错误 #

你遇到的TypeError是因为GraphFrame构造函数必须传入两个参数：顶点DataFrame和边DataFrame，你只传了一个。另外你把边列表DataFrame命名为nodes容易混淆，先修正命名，再生成合规的顶点DataFrame：

步骤1：修正边DataFrame命名 #

将加载的边列表DataFrame改名为edges_df，避免和顶点概念混淆：

schema = StructType([
    StructField("src", IntegerType(), True),
    StructField("dest", IntegerType(), True)])

edges_df = spark.read.format("csv")\
.option("header", "false")\
.schema(schema)\
.option("sep", "\t")\
.load("/content/data/edgelist.txt")
edges_df.show()

步骤2：生成顶点DataFrame #

GraphFrame要求顶点DataFrame必须包含名为id的列，存储所有唯一节点。我们可以从边的src和dest列中提取所有唯一值：

# 提取所有源节点并重命名为id
src_nodes = edges_df.select("src").withColumnRenamed("src", "id")
# 提取所有目标节点并重命名为id
dest_nodes = edges_df.select("dest").withColumnRenamed("dest", "id")
# 合并去重得到完整顶点列表
vertices_df = src_nodes.union(dest_nodes).distinct()

步骤3：正确初始化GraphFrame #

现在用顶点和边DataFrame完成初始化：

from graphframes import GraphFrame

g = GraphFrame(vertices_df, edges_df)
# 验证图结构
print("顶点总数:", g.vertices.count())
print("边总数:", g.edges.count())

问题2：NetworkX可视化的列名错误 #

你的PlotGraph函数里用了select('src','dst')，但实际边DataFrame的目标节点列名是dest不是dst，这会导致列不存在的错误。同时需要添加plt.show()才能显示图像：

import networkx as nx
import matplotlib.pyplot as plt

def PlotGraph(edge_list):
    Gplot = nx.Graph()
    # 修正列名为src和dest
    for row in edge_list.select('src','dest').take(1000):
        Gplot.add_edge(row['src'], row['dest'])
    nx.draw(Gplot, with_labels=True, font_weight='bold')
    plt.show()

# 调用可视化函数
PlotGraph(g.edges)

完整修正代码 #

from pyspark.sql.types import StructType, StructField, IntegerType
from graphframes import GraphFrame
import networkx as nx
import matplotlib.pyplot as plt

# 1. 加载边列表DataFrame
schema = StructType([
    StructField("src", IntegerType(), True),
    StructField("dest", IntegerType(), True)])

edges_df = spark.read.format("csv")\
.option("header", "false")\
.schema(schema)\
.option("sep", "\t")\
.load("/content/data/edgelist.txt")

# 2. 生成顶点DataFrame
src_nodes = edges_df.select("src").withColumnRenamed("src", "id")
dest_nodes = edges_df.select("dest").withColumnRenamed("dest", "id")
vertices_df = src_nodes.union(dest_nodes).distinct()

# 3. 初始化GraphFrame
g = GraphFrame(vertices_df, edges_df)

# 4. 图可视化函数
def PlotGraph(edge_list):
    Gplot = nx.Graph()
    for row in edge_list.select('src','dest').take(1000):
        Gplot.add_edge(row['src'], row['dest'])
    nx.draw(Gplot, with_labels=True, font_weight='bold')
    plt.show()

# 执行可视化
PlotGraph(g.edges)

内容的提问来源于stack exchange，提问作者user21474052