如何从RDD中按行指定元素创建键值对，统计访问地址的出现频次

问题根因 #

• 你使用flatMap处理行数据时，把每行拆分后的所有字段都摊平为了RDD的独立元素，导致最终nasa RDD里混杂了访问地址、请求方法、状态码、返回大小等所有内容，没有区分字段类型，统计时自然会对所有内容计数
• 你编写的process_row函数存在语法和逻辑错误：第一个return words会直接返回整个拆分后的列表，后续返回指定字段的代码永远不会执行；列表取值应该用方括号[]而不是圆括号()；你写的替换逻辑会在最后一个元素为-时把整行所有字段都替换为0，不符合预期。

修正方案 #

第一步：调整RDD转换逻辑，把每行处理为结构化元组 #

不要用flatMap，改用map，让每行处理后对应一个固定结构的元组，第一个元素就是你要统计的访问地址：

def process_row(row):
    words = row.replace('"', '').split()
    # 处理最后一个字段为-的情况
    if words[-1] == '-':
        words[-1] = '0'
    # 按位置返回对应字段，顺序为：访问地址、请求时间、请求方法、请求路径、状态码、返回大小
    return (words[0], words[1], words[2], words[3], words[4], int(words[5]))

# 用map而不是flatMap，保留每行的结构化信息
nasa = nasa_raw.map(process_row)
nasa.persist()

第二步：统计时明确取访问地址字段作为键 #

统计时只取元组的第一个元素（对应访问地址）作为聚合的键即可：

# 取第一个元素作为key，计数1
pairs = nasa.map(lambda x: (x[0], 1))
count_by_address = pairs.reduceByKey(lambda x, y: x + y)
# 取访问频次前10的地址
count_by_address_top10 = count_by_address.takeOrdered(10, key = lambda x: -x[1])
# 输出结果
spark.createDataFrame(count_by_address_top10, ['访问地址','访问频次']).show(10)

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