PySpark：拆分组合键及MAC地址访问目标IP次数统计

嘿，你这个思路完全没问题！用MAC地址+目标IP的组合作为Key，再通过reduceByKey累加计数，正是处理这类访问次数统计的标准操作，我帮你把代码补全并梳理清楚细节：

# 定义映射函数：将原RDD元素转换为((MAC, 目标IP), 1)的键值对
def processJson(data):
    # 假设data是原RDD的单个元素，结构为(MAC_address, dst_ip_address, 1)
    MAC_address = data[0]
    dst_ip_address = data[1]
    return ((MAC_address, dst_ip_address), 1)

# 定义聚合函数：对相同Key的访问次数进行累加
def countreducer(a, b):
    return a + b

# 执行映射与聚合操作
# 注意：如果df是Spark DataFrame，需要先转成RDD，写法为df.rdd.map(...)
count_rdd = df.map(processJson).reduceByKey(countreducer)

# 筛选特定MAC与目标IP的统计结果（按需使用）
target_mac = "AA:BB:CC:DD:EE:FF"
target_ip = "192.168.1.100"
specific_result = count_rdd.filter(lambda x: x[0][0] == target_mac and x[0][1] == target_ip).collect()

# 输出结果
if specific_result:
    print(f"MAC地址 {target_mac} 访问目标IP {target_ip} 的总次数: {specific_result[0][1]}")
else:
    print(f"未找到MAC地址 {target_mac} 访问目标IP {target_ip} 的记录")

关键逻辑说明 #

• 映射阶段：processJson把原数据转换成键值对结构，让每一次访问行为都对应一个((MAC, IP), 1)的条目，这样相同的访问组合会被归为同一个Key。
• 聚合阶段：reduceByKey(countreducer)会自动将相同Key对应的Value值累加，最终得到每个(MAC, IP)组合的总访问次数，这一步是分布式执行的，效率很高。
• 筛选特定结果：如果只需要某一个MAC对某一个IP的统计数据，用filter就能快速定位，最后通过collect()把结果拉取到Driver端进行展示。

另外补充个小建议：如果你的df是Spark DataFrame而不是原生RDD，其实也可以用更简洁的DataFrame API实现相同需求：

result_df = df.groupBy("MAC_address", "dst_ip_address").count()
# 筛选特定结果
specific_df = result_df.filter((result_df.MAC_address == target_mac) & (result_df.dst_ip_address == target_ip))

两种方式各有优劣，RDD更适合自定义复杂逻辑，DataFrame则写法更简洁且有优化器加持~

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