如何在PySpark中高效统计会话中页面对的共现次数？

问题背景 #

你现在需要统计指定页面对在会话中的共现次数，但原来的三重嵌套循环效率极低——哪怕小样本都慢得离谱。你的输入数据如下：

list1 = ['page_a','page_a','page_a','page_b','page_b','page_c']
list2 = ['page_b','page_c','page_d','page_c','page_d','page_d']
sessions = ['page_a,page_b', 'page_b,page_d,page_c', 'page_b', 'page_d,page_c,page_a,page_b,', ]

原代码用了三层循环，时间复杂度是O(MNK)（M是list1长度，N是会话数，K是单会话页面数），完全没法扩展。你期望输出[2,1,1,2,2,2]，同时想用PySpark做并行化处理。

方案一：原生Python优化（小数据场景） #

核心思路是预处理会话为集合，把in操作从O(K)降到O(1)，同时避免重复解析会话字符串：

# 第一步：预处理所有会话，转成去重的页面集合，清理无效空字符串
processed_sessions = []
for s in sessions:
    # 分割并过滤掉空字符串（比如最后一个会话末尾的逗号导致的空值）
    pages = [p.strip() for p in s.split(',') if p.strip()]
    processed_sessions.append(set(pages))

# 第二步：把list1和list2配对
page_pairs = list(zip(list1, list2))

# 第三步：统计每个页面对的共现次数
cooccurrence = [
    sum(1 for session_set in processed_sessions if p1 in session_set and p2 in session_set)
    for p1, p2 in page_pairs
]

print(cooccurrence)  # 输出: [2, 1, 1, 2, 2, 2]

这个优化把时间复杂度降到O(M*N)，比原代码快很多——尤其是当会话页面数较多时，提升效果更明显。如果还想更快，可以用multiprocessing做本地并行，但小数据场景下上面的代码足够了。

方案二：PySpark并行化处理（大数据场景） #

如果你的会话数据量极大，PySpark的分布式计算能完美解决效率问题。步骤如下：

from pyspark.sql import SparkSession
from pyspark.sql.functions import udf, explode, count
from pyspark.sql.types import ArrayType, StringType, StructType, StructField

# 1. 初始化SparkSession
spark = SparkSession.builder.appName("PageCooccurrence").getOrCreate()

# 2. 预处理会话数据：清理空值，转成数组（Spark不直接支持集合类型）
session_records = []
for session_id, s in enumerate(sessions):
    pages = [p.strip() for p in s.split(',') if p.strip()]
    session_records.append((str(session_id), pages))

# 3. 创建会话DataFrame
session_schema = StructType([
    StructField("session_id", StringType(), nullable=False),
    StructField("pages", ArrayType(StringType()), nullable=False)
])
session_df = spark.createDataFrame(session_records, schema=session_schema)

# 4. 准备带索引的页面对（方便最后还原原顺序）
pair_with_index = [
    (str(idx), p1, p2) for idx, (p1, p2) in enumerate(zip(list1, list2))
]
pair_schema = StructType([
    StructField("pair_idx", StringType(), nullable=False),
    StructField("page1", StringType(), nullable=False),
    StructField("page2", StringType(), nullable=False)
])
pair_df = spark.createDataFrame(pair_with_index, schema=pair_schema)

# 5. 广播页面对列表（减少节点间数据传输，提升效率）
broadcast_pairs = spark.sparkContext.broadcast(pair_with_index)

# 6. 定义UDF：找出当前会话包含的所有目标页面对的索引
def find_matching_pairs(page_list):
    page_set = set(page_list)
    matching_indices = []
    for idx, p1, p2 in broadcast_pairs.value:
        if p1 in page_set and p2 in page_set:
            matching_indices.append(idx)
    return matching_indices

find_matching_pairs_udf = udf(find_matching_pairs, ArrayType(StringType()))

# 7. 计算每个页面对的共现次数
result_df = session_df.withColumn("matching_pair_ids", find_matching_pairs_udf("pages")) \
    .select(explode("matching_pair_ids").alias("pair_idx")) \
    .groupBy("pair_idx") \
    .agg(count("*").alias("cooccur_count"))

# 8. 还原原页面对的顺序（确保共现0次的对也被保留）
base_df = spark.createDataFrame([(str(idx),) for idx in range(len(pair_with_index))], ["pair_idx"])
final_df = base_df.join(result_df, on="pair_idx", how="left") \
    .fillna(0, subset=["cooccur_count"]) \
    .orderBy("pair_idx") \
    .select("cooccur_count")

# 9. 获取最终结果列表
cooccurrence_spark = [row.cooccur_count for row in final_df.collect()]
print(cooccurrence_spark)  # 输出: [2, 1, 1, 2, 2, 2]

# 关闭SparkSession
spark.stop()

关键优化点：

• 用广播变量把页面对列表分发到所有节点，避免重复传输大对象
• 每个节点并行处理部分会话，把计算压力分散到集群
• 用left join确保所有原页面对都被统计（包括共现次数为0的）

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