这次的作业主要用PySpark来分析Web Server Log。主要分成4个部分。相关ipynb文件见我github。

Part 1 Apache Web Server Log file format

这部分主要是了解log file的格式，然后处理它。我们处理的日志格式符合Common Log Format（CLF）标准。其一行记录长这样：

127.0.0.1 - - [01/Aug/1995:00:00:01 -0400] "GET /images/launch-logo.gif HTTP/1.0" 200 1839

这次的作业数据来源于NASA Kennedy Space Center WWW server。其完整的数据在这个网址可以免费得到（http://ita.ee.lbl.gov/html/contrib/NASA-HTTP.html）

Parsing Each Log Line

我们的第一步当然是解析数据，从原始数据中选出有用的数据来。不过这段代码作业为我们准备好了。

import re
import datetimefrom pyspark.sql import Rowmonth_map = {'Jan': 1, 'Feb': 2, 'Mar':3, 'Apr':4, 'May':5, 'Jun':6, 'Jul':7,'Aug':8,  'Sep': 9, 'Oct':10, 'Nov': 11, 'Dec': 12}def parse_apache_time(s):""" Convert Apache time format into a Python datetime objectArgs:s (str): date and time in Apache time formatReturns:datetime: datetime object (ignore timezone for now)"""return datetime.datetime(int(s[7:11]),month_map[s[3:6]],int(s[0:2]),int(s[12:14]),int(s[15:17]),int(s[18:20]))def parseApacheLogLine(logline):""" Parse a line in the Apache Common Log formatArgs:logline (str): a line of text in the Apache Common Log formatReturns:tuple: either a dictionary containing the parts of the Apache Access Log and 1,or the original invalid log line and 0"""match = re.search(APACHE_ACCESS_LOG_PATTERN, logline)if match is None:return (logline, 0)size_field = match.group(9)if size_field == '-':size = long(0)else:size = long(match.group(9))return (Row(host          = match.group(1),client_identd = match.group(2),user_id       = match.group(3),date_time     = parse_apache_time(match.group(4)),method        = match.group(5),endpoint      = match.group(6),protocol      = match.group(7),response_code = int(match.group(8)),content_size  = size), 1)

# A regular expression pattern to extract fields from the log line
APACHE_ACCESS_LOG_PATTERN = '^(\S+) (\S+) (\S+) \[([\w:/]+\s[+\-]\d{4})\] "(\S+) (\S+)\s*(\S*)" (\d{3}) (\S+)'

Configuration and Initial RDD Creation

import sys
import os
from test_helper import TestbaseDir = os.path.join('data')
inputPath = os.path.join('cs100', 'lab2', 'apache.access.log.PROJECT')
logFile = os.path.join(baseDir, inputPath)def parseLogs():""" Read and parse log file """parsed_logs = (sc.textFile(logFile).map(parseApacheLogLine).cache())access_logs = (parsed_logs.filter(lambda s: s[1] == 1).map(lambda s: s[0]).cache())failed_logs = (parsed_logs.filter(lambda s: s[1] == 0).map(lambda s: s[0]))failed_logs_count = failed_logs.count()if failed_logs_count > 0:print 'Number of invalid logline: %d' % failed_logs.count()for line in failed_logs.take(20):print 'Invalid logline: %s' % lineprint 'Read %d lines, successfully parsed %d lines, failed to parse %d lines' % (parsed_logs.count(), access_logs.count(), failed_logs.count())return parsed_logs, access_logs, failed_logsparsed_logs, access_logs, failed_logs = parseLogs()

这段代码是把log file数据转换成RDD，然后用上一节里的解析函数来解析，最后把结果缓存起来。因为后面还需要用这个结果。通过上面的步骤，我们发现，有大量的记录无法解析。

Number of invalid logline: 108
Invalid logline: ix-sac6-20.ix.netcom.com - - [08/Aug/1995:14:43:39 -0400] "GET / HTTP/1.0 " 200 7131
Invalid logline: ix-sac6-20.ix.netcom.com - - [08/Aug/1995:14:43:57 -0400] "GET /images/ksclogo-medium.gif HTTP/1.0 " 200 5866
Invalid logline: ix-sac6-20.ix.netcom.com - - [08/Aug/1995:14:44:07 -0400] "GET /images/NASA-logosmall.gif HTTP/1.0 " 200 786
Invalid logline: ix-sac6-20.ix.netcom.com - - [08/Aug/1995:14:44:11 -0400] "GET /images/MOSAIC-logosmall.gif HTTP/1.0 " 200 363
Invalid logline: ix-sac6-20.ix.netcom.com - - [08/Aug/1995:14:44:13 -0400] "GET /images/USA-logosmall.gif HTTP/1.0 " 200 234
Invalid logline: ix-sac6-20.ix.netcom.com - - [08/Aug/1995:14:44:15 -0400] "GET /images/WORLD-logosmall.gif HTTP/1.0 " 200 669
Invalid logline: ix-sac6-20.ix.netcom.com - - [08/Aug/1995:14:44:31 -0400] "GET /shuttle/countdown/ HTTP/1.0 " 200 4673
Invalid logline: ix-sac6-20.ix.netcom.com - - [08/Aug/1995:14:44:41 -0400] "GET /shuttle/missions/sts-69/count69.gif HTTP/1.0 " 200 46053
Invalid logline: ix-sac6-20.ix.netcom.com - - [08/Aug/1995:14:45:34 -0400] "GET /images/KSC-logosmall.gif HTTP/1.0 " 200 1204
Invalid logline: ix-sac6-20.ix.netcom.com - - [08/Aug/1995:14:45:46 -0400] "GET /cgi-bin/imagemap/countdown69?293,287 HTTP/1.0 " 302 85
Invalid logline: ix-sac6-20.ix.netcom.com - - [08/Aug/1995:14:45:48 -0400] "GET /htbin/cdt_main.pl HTTP/1.0 " 200 3714
Invalid logline: ix-sac6-20.ix.netcom.com - - [08/Aug/1995:14:45:52 -0400] "GET /shuttle/countdown/images/countclock.gif HTTP/1.0 " 200 13994
Invalid logline: ix-li1-14.ix.netcom.com - - [08/Aug/1995:14:46:22 -0400] "GET / HTTP/1.0 " 200 7131
Invalid logline: ix-li1-14.ix.netcom.com - - [08/Aug/1995:14:46:29 -0400] "GET /images/ksclogo-medium.gif HTTP/1.0 " 200 5866
Invalid logline: ix-li1-14.ix.netcom.com - - [08/Aug/1995:14:46:35 -0400] "GET /images/NASA-logosmall.gif HTTP/1.0 " 200 786
Invalid logline: ix-li1-14.ix.netcom.com - - [08/Aug/1995:14:46:37 -0400] "GET /images/MOSAIC-logosmall.gif HTTP/1.0 " 200 363
Invalid logline: ix-li1-14.ix.netcom.com - - [08/Aug/1995:14:46:38 -0400] "GET /images/USA-logosmall.gif HTTP/1.0 " 200 234
Invalid logline: ix-li1-14.ix.netcom.com - - [08/Aug/1995:14:46:40 -0400] "GET /images/WORLD-logosmall.gif HTTP/1.0 " 200 669
Invalid logline: ix-li1-14.ix.netcom.com - - [08/Aug/1995:14:47:41 -0400] "GET /shuttle/missions/sts-70/mission-sts-70.html HTTP/1.0 " 200 20304
Invalid logline: ix-sac6-20.ix.netcom.com - - [08/Aug/1995:14:47:48 -0400] "GET /shuttle/countdown/count.html HTTP/1.0 " 200 73231
Read 1043177 lines, successfully parsed 1043069 lines, failed to parse 108 lines

Data Cleaning

我们发现有108行解析失败。所以我们要重新写正则表达式以此来把数据全部解析成功。

# TODO: Replace <FILL IN> with appropriate code# This was originally '^(\S+) (\S+) (\S+) \[([\w:/]+\s[+\-]\d{4})\] "(\S+) (\S+)\s*(\S*)" (\d{3}) (\S+)'
APACHE_ACCESS_LOG_PATTERN = '^(\S+) (\S+) (\S+) \[([\w:/]+\s[+\-]\d{4})\] "(\S+) (\S+)\s*(\S*).*" (\d{3}) (\S+)'parsed_logs, access_logs, failed_logs = parseLogs()

这个规则能把日志文件全部解析出来。

Part 2 Sample Analyses on the Web Server Log File

Example: Content Size Statistics

这里的操作就是一些常见的统计分析，出现了两个之前没出现的方法min()和max()。

# Calculate statistics based on the content size.
content_sizes = access_logs.map(lambda log: log.content_size).cache()
print 'Content Size Avg: %i, Min: %i, Max: %s' % (content_sizes.reduce(lambda a, b : a + b) / content_sizes.count(),content_sizes.min(),content_sizes.max())

Example: Response Code Analysis

这里是分析response code，主要是统计有几种不同的response code和各种有多少个。

# Response Code to Count
responseCodeToCount = (access_logs.map(lambda log: (log.response_code, 1)).reduceByKey(lambda a, b : a + b).cache())
responseCodeToCountList = responseCodeToCount.take(100)
print 'Found %d response codes' % len(responseCodeToCountList)
print 'Response Code Counts: %s' % responseCodeToCountList
assert len(responseCodeToCountList) == 7
assert sorted(responseCodeToCountList) == [(200, 940847), (302, 16244), (304, 79824), (403, 58), (404, 6185), (500, 2), (501, 17)]

Example: Response Code Graphing with matplotlib

这里主要是在上面的结果基础上，算每个code的百分比，画出其扇形百分比的图。

labels = responseCodeToCount.map(lambda (x, y): x).collect()
print labels
count = access_logs.count()
fracs = responseCodeToCount.map(lambda (x, y): (float(y) / count)).collect()
print fracs

import matplotlib.pyplot as pltdef pie_pct_format(value):""" Determine the appropriate format string for the pie chart percentage labelArgs:value: value of the pie sliceReturns:str: formated string label; if the slice is too small to fit, returns an empty string for label"""return '' if value < 7 else '%.0f%%' % valuefig = plt.figure(figsize=(4.5, 4.5), facecolor='white', edgecolor='white')
colors = ['yellowgreen', 'lightskyblue', 'gold', 'purple', 'lightcoral', 'yellow', 'black']
explode = (0.05, 0.05, 0.1, 0, 0, 0, 0)
patches, texts, autotexts = plt.pie(fracs, labels=labels, colors=colors,explode=explode, autopct=pie_pct_format,shadow=False,  startangle=125)
for text, autotext in zip(texts, autotexts):if autotext.get_text() == '':text.set_text('')  # If the slice is small to fit, don't show a text label
plt.legend(labels, loc=(0.80, -0.1), shadow=True)
pass

Example: Frequent Hosts

这里是研究host，统计每个host出现的次数，选出出现次数超过十次的host（其实和前面的工作差不多）

# Any hosts that has accessed the server more than 10 times.
hostCountPairTuple = access_logs.map(lambda log: (log.host, 1))hostSum = hostCountPairTuple.reduceByKey(lambda a, b : a + b)hostMoreThan10 = hostSum.filter(lambda s: s[1] > 10)hostsPick20 = (hostMoreThan10.map(lambda s: s[0]).take(20))print 'Any 20 hosts that have accessed more then 10 times: %s' % hostsPick20
# An example: [u'204.120.34.185', u'204.243.249.9', u'slip1-32.acs.ohio-state.edu', u'lapdog-14.baylor.edu', u'199.77.67.3', u'gs1.cs.ttu.edu', u'haskell.limbex.com', u'alfred.uib.no', u'146.129.66.31', u'manaus.bologna.maraut.it', u'dialup98-110.swipnet.se', u'slip-ppp02.feldspar.com', u'ad03-053.compuserve.com', u'srawlin.opsys.nwa.com', u'199.202.200.52', u'ix-den7-23.ix.netcom.com', u'151.99.247.114', u'w20-575-104.mit.edu', u'205.25.227.20', u'ns.rmc.com']

Example: Visualizing Endpoints

这次是研究endpoints。。和上面几乎一样。

endpoints = (access_logs.map(lambda log: (log.endpoint, 1)).reduceByKey(lambda a, b : a + b).cache())
ends = endpoints.map(lambda (x, y): x).collect()
counts = endpoints.map(lambda (x, y): y).collect()
fig = plt.figure(figsize=(8,4.2), facecolor='white', edgecolor='white')
plt.axis([0, len(ends), 0, max(counts)])
plt.grid(b=True, which='major', axis='y')
plt.xlabel('Endpoints')
plt.ylabel('Number of Hits')
plt.plot(counts)
pass

Example: Top Endpoints

在上面的结果基础上，排序，选出前十个。

# Top Endpoints
endpointCounts = (access_logs.map(lambda log: (log.endpoint, 1)).reduceByKey(lambda a, b : a + b))topEndpoints = endpointCounts.takeOrdered(10, lambda s: -1 * s[1])print 'Top Ten Endpoints: %s' % topEndpoints
assert topEndpoints == [(u'/images/NASA-logosmall.gif', 59737), (u'/images/KSC-logosmall.gif', 50452), (u'/images/MOSAIC-logosmall.gif', 43890), (u'/images/USA-logosmall.gif', 43664), (u'/images/WORLD-logosmall.gif', 43277), (u'/images/ksclogo-medium.gif', 41336), (u'/ksc.html', 28582), (u'/history/apollo/images/apollo-logo1.gif', 26778), (u'/images/launch-logo.gif', 24755), (u'/', 20292)], 'incorrect Top Ten Endpoints'

Part 3 Analyzing Web Server Log File

part2几乎没有自己写代码。这部分开始，主要是我们写代码实现要求。

Top Ten Error Endpoints

这个要求是返回排名前10的endpoints且response code不是200.我们需要考虑实现的步骤。

1. 过滤掉response code为200的记录
2. 统计每个endpoint的记录
3. 选出排名前十的endpiont

# TODO: Replace <FILL IN> with appropriate code
# HINT: Each of these <FILL IN> below could be completed with a single transformation or action.
# You are welcome to structure your solution in a different way, so long as
# you ensure the variables used in the next Test section are defined (ie. endpointSum, topTenErrURLs).not200 = access_logs.map(lambda log: (log.endpoint, log.response_code)).filter(lambda x:x[1] != 200)endpointCountPairTuple = not200.map(lambda x: (x[0], 1))endpointSum = endpointCountPairTuple.reduceByKey(lambda a, b : a + b)topTenErrURLs = endpointSum.takeOrdered(10, lambda s: -1 * s[1])
print 'Top Ten failed URLs: %s' % topTenErrURLs

Number of Unique Hosts

统计有多少个不同的hosts。这个太简单了：提取host，distinct()后直接count()

# TODO: Replace <FILL IN> with appropriate code
# HINT: Do you recall the tips from (3a)? Each of these <FILL IN> could be an transformation or action.hosts = access_logs.map(lambda log:log.host)uniqueHosts = hosts.distinct()uniqueHostCount = uniqueHosts.count()
print 'Unique hosts: %d' % uniqueHostCount

Number of Unique Daily Hosts

这里要统计每天出现的unique hosts数量，要求假定在了一个月内，所以我们只需要提取days的信息就行，最后别忘了把结果缓存起来。

# TODO: Replace <FILL IN> with appropriate codedayToHostPairTuple = access_logs.map(lambda log:(log.date_time.day,log.host)).distinct()dayGroupedHosts = dayToHostPairTuple.groupByKey()dayHostCount = dayGroupedHosts.mapValues(len)dailyHosts = (dayHostCount.sortByKey().cache())dailyHostsList = dailyHosts.take(30)
print 'Unique hosts per day: %s' % dailyHostsList

Visualizing the Number of Unique Daily Hosts

把上面的结果画图，这里说一下的就是，因为我们用的是list来画图，稍微注意前面代码的人就应该知道collect()可以把RDD转为list。

# TODO: Replace <FILL IN> with appropriate codedaysWithHosts = dailyHosts.map(lambda x : x[0]).collect()
hosts = dailyHosts.map(lambda x : x[1]).collect()
fig = plt.figure(figsize=(8,4.5), facecolor='white', edgecolor='white')
plt.axis([min(daysWithHosts), max(daysWithHosts), 0, max(hosts)+500])
plt.grid(b=True, which='major', axis='y')
plt.xlabel('Day')
plt.ylabel('Hosts')
plt.plot(daysWithHosts, hosts)
pass

Average Number of Daily Requests per Hosts

这次我们要计算每天平均每个host有多少次请求。所以我们要用到join()了，把每天的host数目和每天request数据放到一个tuple里，然后相除得到均值。其中每天host数目我们算好了，即dailyHosts，这里需要算每天的请求数，其实就是host的总数。join后的结果大概是(key, (request, host))这样的结构。

# TODO: Replace <FILL IN> with appropriate codedayAndHostTuple = access_logs.map(lambda log:(log.date_time.day,log.host))groupedByDay = dayAndHostTuple.groupByKey()sortedByDay = groupedByDay.mapValues(len).join(dailyHosts)avgDailyReqPerHost = sortedByDay.map(lambda x :(x[0], x[1][0]/x[1][1])).sortByKey().cache() avgDailyReqPerHostList = avgDailyReqPerHost.take(30)
print 'Average number of daily requests per Hosts is %s' % avgDailyReqPerHostList

Visualizing the Average Daily Requests per Unique Host

又是可视化。。。

# TODO: Replace <FILL IN> with appropriate codedaysWithAvg = avgDailyReqPerHost.map(lambda x : x[0]).collect()
avgs = avgDailyReqPerHost.map(lambda x : x[1]).collect()

fig = plt.figure(figsize=(8,4.2), facecolor='white', edgecolor='white')
plt.axis([0, max(daysWithAvg), 0, max(avgs)+2])
plt.grid(b=True, which='major', axis='y')
plt.xlabel('Day')
plt.ylabel('Average')
plt.plot(daysWithAvg, avgs)
pass

Part 4 Exploring 404 Response Codes

这次主要研究response code为404的记录。

Counting 404 Response Codes

统计记录条数

# TODO: Replace <FILL IN> with appropriate codebadRecords = access_logs.filter(lambda log: log.response_code== 404).cache()print 'Found %d 404 URLs' % badRecords.count()

Listing 404 Response Code Records

# TODO: Replace <FILL IN> with appropriate codebadEndpoints = badRecords.map(lambda log: log.endpoint)badUniqueEndpoints = badEndpoints.distinct()badUniqueEndpointsPick40 = badUniqueEndpoints.take(40)
print '404 URLS: %s' % badUniqueEndpointsPick40

这个也没什么意思，就瞅瞅返回404的url长啥样。

Listing the Top Twenty 404 Response Code Endpoints

# TODO: Replace <FILL IN> with appropriate codebadEndpointsCountPairTuple = badRecords.map(lambda log: (log.endpoint,1))badEndpointsSum = badEndpointsCountPairTuple.reduceByKey(lambda a,b: a+b)badEndpointsTop20 = badEndpointsSum.takeOrdered(20,key=lambda x: -x[1])
print 'Top Twenty 404 URLs: %s' % badEndpointsTop20

排序问题，碰到好多次了。

Listing the Top Twenty-five 404 Response Code Hosts

# TODO: Replace <FILL IN> with appropriate codeerrHostsCountPairTuple = badRecords.map(lambda log: (log.host,1))errHostsSum = errHostsCountPairTuple.reduceByKey(lambda a,b: a+b)errHostsTop25 = errHostsSum.takeOrdered(25,key=lambda x: -x[1])
print 'Top 25 hosts that generated errors: %s' % errHostsTop25

同上，只是把endpoint改成了host（你们是不是发现pyspark真的很简单。。。）

Listing 404 Response Codes per Day

# TODO: Replace <FILL IN> with appropriate codeerrDateCountPairTuple = badRecords.map(lambda log:(log.date_time.day,1))errDateSum = errDateCountPairTuple.reduceByKey(lambda a,b : a+b)errDateSorted = errDateSum.sortByKey().cache()errByDate = errDateSorted.collect()
print '404 Errors by day: %s' % errByDate

统计每天的记录数目

Visualizing the 404 Response Codes by Day

# TODO: Replace <FILL IN> with appropriate codedaysWithErrors404 = errDateSorted.map(lambda x :x[0]).collect()
errors404ByDay = errDateSorted.map(lambda x :x[1]).collect()

fig = plt.figure(figsize=(8,4.2), facecolor='white', edgecolor='white')
plt.axis([0, max(daysWithErrors404), 0, max(errors404ByDay)])
plt.grid(b=True, which='major', axis='y')
plt.xlabel('Day')
plt.ylabel('404 Errors')
plt.plot(daysWithErrors404, errors404ByDay)
pass

Top Five Days for 404 Response Codes

# TODO: Replace <FILL IN> with appropriate codetopErrDate = errDateSorted.takeOrdered(5,key=lambda x: -x[1])
print 'Top Five dates for 404 requests: %s' % topErrDate

又是top n的问题。。

Hourly 404 Response Codes

# TODO: Replace <FILL IN> with appropriate codehourCountPairTuple = badRecords.map(lambda log:(log.date_time.hour,log.response_code))hourRecordsSum = hourCountPairTuple.groupByKey().mapValues(len)hourRecordsSorted = hourRecordsSum.sortByKey().cache()errHourList = hourRecordsSorted.collect()
print 'Top hours for 404 requests: %s' % errHourList

和前面几乎一样，只不过把天换成了小时。

Visualizing the 404 Response Codes by Hour

老朋友了，可视化。。

# TODO: Replace <FILL IN> with appropriate codehoursWithErrors404 = hourRecordsSorted.map(lambda x :x[0]).collect()
errors404ByHours = hourRecordsSorted.map(lambda x :x[1]).collect()

fig = plt.figure(figsize=(8,4.2), facecolor='white', edgecolor='white')
plt.axis([0, max(hoursWithErrors404), 0, max(errors404ByHours)])
plt.grid(b=True, which='major', axis='y')
plt.xlabel('Hour')
plt.ylabel('404 Errors')
plt.plot(hoursWithErrors404, errors404ByHours)
pass

如果有任何问题，可以去我的github上下载源文件自己做一做，欢迎交流。