52 lines
1.5 KiB
Python
52 lines
1.5 KiB
Python
import re
|
||
from pyspark import SparkContext
|
||
|
||
sc = SparkContext.getOrCreate()
|
||
|
||
logFile = "hdfs://master:9000/user/root/apache.access.log.PROJECT"
|
||
raw_logs = sc.textFile(logFile)
|
||
|
||
LOG_PATTERN = re.compile(r'^(\S+) (\S+) (\S+) \[([\w:/]+\s[+-]\d{4})\] "(\S+) (\S+)\s*(\S*)\s?" (\d{3}) (\S+)')
|
||
|
||
# 解析日期和主机
|
||
def parse_day_ip(line):
|
||
match = LOG_PATTERN.match(line)
|
||
if not match:
|
||
return None
|
||
ip = match.group(1)
|
||
full_time = match.group(4)
|
||
day = full_time.split('/')[0] # "10"
|
||
return (day, ip)
|
||
|
||
# 1️⃣ 每天总请求数
|
||
def parse_day(line):
|
||
match = LOG_PATTERN.match(line)
|
||
if not match:
|
||
return None
|
||
full_time = match.group(4)
|
||
day = full_time.split('/')[0]
|
||
return (day, 1)
|
||
|
||
# 得到每天总请求数
|
||
dailyRequests = raw_logs.map(parse_day).filter(lambda x: x is not None) \
|
||
.reduceByKey(lambda a, b: a + b)
|
||
|
||
# 得到每天不同主机数（如 1.3 中的逻辑）
|
||
day_ip_pairs = raw_logs.map(parse_day_ip).filter(lambda x: x is not None)
|
||
dailyUniqueHosts = day_ip_pairs.distinct() \
|
||
.map(lambda x: (x[0], 1)) \
|
||
.reduceByKey(lambda a, b: a + b)
|
||
|
||
# 2️⃣ 合并两个 RDD，并计算平均值（用整数除法）
|
||
dailyReqJoinHost = dailyRequests.join(dailyUniqueHosts)
|
||
|
||
avgDailyReqPerHost = dailyReqJoinHost.map(
|
||
lambda x: (x[0], x[1][0] // x[1][1]) # (day, total_requests // unique_hosts)
|
||
).sortByKey().cache() # ⚠️ 缓存结果供后续使用
|
||
|
||
# 输出看看
|
||
for day, avg in avgDailyReqPerHost.collect():
|
||
print("Day {}: 平均每个主机请求 {} 次".format(day, avg))
|
||
|
||
sc.stop()
|