如何在PySpark中统计带权DAG的所有节点总数？

要同时统计制表符左侧的节点和右侧邻接字典中的所有节点，只需在单个处理流程中同时提取两类节点，再统一去重计数即可。以下是修改后的完整代码：

import ast

def extract_all_nodes(line):
    # 拆分当前节点与邻接字典字符串
    node_str, adj_dict_str = line.split('\t')
    current_node = int(node_str)
    # 解析右侧的邻接字典
    adjacency_dict = ast.literal_eval(adj_dict_str)
    
    # 先返回当前节点
    yield current_node
    # 再逐个返回邻接字典中的所有目标节点
    for target_node_key in adjacency_dict.keys():
        yield int(target_node_key)

# 处理RDD，提取所有节点并去重
all_unique_nodes = dataRDD.flatMap(extract_all_nodes).distinct()
# 统计总节点数
total_node_count = all_unique_nodes.count()
print(total_node_count)  # 输出结果为11，与手动统计一致

代码逻辑说明 #

1. 单行数据处理：extract_all_nodes函数会对每一行数据做两件事：
   • 拆分出左侧的当前节点ID，转换为整数；
   • 解析右侧的邻接字典字符串，提取所有目标节点ID并转换为整数。
2. 扁平化输出：通过flatMap将每个行生成的节点列表展开为单个元素的RDD，确保所有节点都被纳入统计范围。
3. 去重与计数：调用distinct()去除重复节点，最后用count()得到所有唯一节点的总数。

这种方式避免了分开处理左右节点再合并的繁琐，一次性覆盖了所有节点（包括仅出现在邻接字典中的节点如1），确保统计结果准确。

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