PySpark RDD统计日志文件服务器名出现次数结果为0如何解决

• 变量引用错误：代码中已定义日志DataFrame为base_df，转换后的RDD为base_df_rdd，但统计逻辑中错误引用了未声明的textFile变量，若实际运行无报错则大概率为编写示例时的笔误，核心问题为匹配逻辑错误。
• 匹配规则不符合场景：x[0].split(' ').count(i)的逻辑是按单个空格拆分整行日志后，统计列表中完全等于指定server_name的元素个数。如果日志中server_name前后存在括号、引号等额外字符，或日志字段分隔符为多空格、制表符等非单空格符号，拆分后的列表不会存在完全匹配的项，最终统计结果为0。
• 潜在闭包绑定问题：Python lambda存在延迟绑定特性，循环中定义的lambda表达式捕获的i变量会在实际执行时才取值，分布式运行场景下可能出现捕获值不符合预期的问题。

方案1：修正RDD实现逻辑 #

无需拆分日志行，直接统计每行中指定server_name的出现次数即可：

from pyspark.context import SparkContext
from pyspark.sql.session import SparkSession
   
sc = SparkContext()
spark = SparkSession(sc)
base_df = spark.read.text("/content/fsm-20210817.logs")
base_rdd = base_df.rdd

server_list = ['nginx-ingress-controller-5b6697898-zqxl4','cert-manager-5695c78d49-q9s9j']

for server_name in server_list:
    count = base_rdd.map(lambda row: row.value.count(server_name)).sum()
    print(server_name, count)

方案2：使用DataFrame API优化性能 #

直接使用Spark原生DataFrame算子实现，避免RDD转换开销，执行效率更高：

from pyspark.sql.functions import col, instr

base_df = spark.read.text("/content/fsm-20210817.logs")
server_list = ['nginx-ingress-controller-5b6697898-zqxl4','cert-manager-5695c78d49-q9s9j']

for server_name in server_list:
    # 统计包含指定server_name的日志行数，若需统计单行多次出现的总次数可替换为正则统计逻辑
    count = base_df.filter(instr(col("value"), server_name) > 0).count()
    print(server_name, count)

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