import re from pyspark import SparkContext import matplotlib.pyplot as plt # 初始化 SparkContext sc = SparkContext.getOrCreate() # 日志匹配的正则表达式 LOG_PATTERN = re.compile( r'^(\S+) (\S+) (\S+) \[([\w:/]+\s[+-]\d{4})\] "(\S+) (\S+)\s*(\S*)\s?" (\d{3}) (\S+)' ) # 解析日志的函数 def parse_log_line(line): match = LOG_PATTERN.match(line) if not match: return None content_size_str = match.group(9) content_size = int(content_size_str) if content_size_str.isdigit() else 0 return { 'ip': match.group(1), 'user_identity': match.group(2), 'user_id': match.group(3), 'timestamp': match.group(4), 'method': match.group(5), 'endpoint': match.group(6), 'protocol': match.group(7), 'status_code': int(match.group(8)), 'content_size': content_size } # 提取小时 def extract_hour(log): timestamp = log['timestamp'] hour = timestamp.split(":")[1] # 从时间戳中提取小时 return hour if __name__ == "__main__": # 加载日志文件 logFile = "hdfs://master:9000/user/root/apache.access.log.PROJECT" raw_logs = sc.textFile(logFile) # 解析并过滤有效日志 access_logs = raw_logs.map(parse_log_line).filter(lambda x: x is not None).cache() # 只保留 404 响应代码的日志 bad_records = access_logs.filter(lambda log: log['status_code'] == 404).cache() # 提取每小时的 404 错误记录 hourly_404_counts = bad_records.map(lambda log: (extract_hour(log), 1)) \ .reduceByKey(lambda a, b: a + b) \ .sortByKey() # 按小时排序 # 将结果转换为列表 hourly_404_counts_list = hourly_404_counts.collect() # 提取小时和对应的 404 次数 hours = [hour for hour, count in hourly_404_counts_list] counts = [count for hour, count in hourly_404_counts_list] # 使用 matplotlib 绘制条形图 plt.figure(figsize=(10, 6)) plt.bar(hours, counts, color='blue') plt.title('每小时404响应代码数量') plt.xlabel('小时') plt.ylabel('404响应次数') plt.xticks(rotation=45) # 将小时标签旋转 45 度 plt.tight_layout() plt.show() # 停止 Spark sc.stop()