Azure HDInsight 上的 Spark 群集配合自定义的Python来分析网站日志

一、前言：本文是个实践博客，演示如何结合使用自定义库和 HDInsight 上的 Spark 来分析日志数据。 我们使用的自定义库是一个名为 iislogparser.py的 Python 库。

1. 每步的输入和对应的输出
2. 纠正了原文中一个因版本引起的小问题

前提：你先在Azure HDinsight上有一个Apache Spark集群，（似乎现在只能是2.*版本的spark了）

二、将原始数据另存为 RDD

在本部分中，将使用与 HDInsight 中的 Apache Spark 群集关联的 Jupyter 笔记本来运行用于处理原始示例数据并将其保存为 Hive 表的作业。 示例数据是所有群集在默认情况下均会提供的 .csv 文件 (hvac.csv)。 将数据保存为 Hive 表后，下一节将使用 Power BI 和 Tableau 等 BI 工具连接 Hive 表。

1. 在 Azure 门户上的启动板中，单击 Spark 群集的磁贴（如果已将它固定到启动板）。 也可以单击“全部浏览” > “HDInsight 群集”导航到群集。

2. 在 Spark 群集边栏选项卡中单击“群集仪表板”，然后单击“Jupyter Notebook”。 出现提示时，请输入群集的管理员凭据。

3. 创建新的笔记本。 单击“新建”，然后单击“PySpark”。

4. 随即创建新笔记本，并以 Untitled.pynb 名称打开。 单击顶部的笔记本名称，并输入一个友好名称。

5. 使用笔记本是使用 PySpark 内核创建的，因此不需要显式创建任何上下文。 运行第一个代码单元格时，系统自动创建 Spark 和 Hive 上下文。 首先，可以导入此方案所需的类型。 将以下代码段粘贴到空白单元格中，并按 SHIFT + ENTER。

from pyspark.sql import Row
from pyspark.sql.types import *

6. 使用群集上已可用的示例日志数据创建 RDD。 可以从 HdiSamplesHdiSamplesWebsiteLogSampleDataSampleLog909f2b.log 中访问与群集关联的默认存储帐户中的数据。

logs = sc.textFile('wasb:///HdiSamples/HdiSamples/WebsiteLogSampleData/SampleLog/909f2b.log')

此处的日志格式大致为：

#Software: Microsoft Internet Information Services 8.0
#Fields: date time s-sitename cs-method cs-uri-stem cs-uri-query s-port cs-username c-ip cs(User-Agent) cs(Cookie) cs(Referer) cs-host sc-status sc-substatus sc-win32-status sc-bytes cs-bytes time-taken
2014-01-01 02:01:09 SAMPLEWEBSITE GET /blogposts/mvc4/step2.png X-ARR-LOG-ID=2ec4b8ad-3cf0-4442-93ab-837317ece6a1 80 - 1.54.23.196 Mozilla/5.0+(Windows+NT+6.3;+WOW64)+AppleWebKit/537.36+(KHTML,+like+Gecko)+Chrome/31.0.1650.63+Safari/537.36 - http://weblogs.asp.net/sample/archive/2007/12/09/asp-net-mvc-framework-part-4-handling-form-edit-and-post-scenarios.aspx www.sample.com 200 0 0 53175 871 46
2014-01-01 02:01:09 SAMPLEWEBSITE GET /blogposts/mvc4/step3.png X-ARR-LOG-ID=9eace870-2f49-4efd-b204-0d170da46b4a 80 - 1.54.23.196 Mozilla/5.0+(Windows+NT+6.3;+WOW64)+AppleWebKit/537.36+(KHTML,+like+Gecko)+Chrome/31.0.1650.63+Safari/537.36 - http://weblogs.asp.net/sample/archive/2007/12/09/asp-net-mvc-framework-part-4-handling-form-edit-and-post-scenarios.aspx www.sample.com 200 0 0 51237 871 32
2014-01-01 02:21:19 SAMPLEWEBSITE GET /blogposts/mvcrouting/step11.jpg X-ARR-LOG-ID=117d64f4-5bf4-44e0-9ef5-669737d69adc 80 - 115.64.147.147 Mozilla/5.0+(Macintosh;+Intel+Mac+OS+X+10_9_1)+AppleWebKit/537.36+(KHTML,+like+Gecko)+Chrome/31.0.1650.63+Safari/537.36 ARRAffinity=909f2bd5d754dbde10a14dff095f3fbe3b22ca6b04cccdaf15bd58ecb51e1fe4;+WAWebSiteSID=7613e9e104a04a31a798554d8954622c http://weblogs.asp.net/sample/archive/2007/12/03/asp-net-mvc-framework-part-2-url-routing.aspx www.sample.com 200 0 0 30944 976 15

7. 检索示例日志集以验证上一步是否成功完成。

logs.take(5)

应该会看到与下面类似的输出：

 [u'#Software: Microsoft Internet Information Services 8.0',
  u'#Fields: date time s-sitename cs-method cs-uri-stem cs-uri-query s-port cs-username c-ip cs(User-Agent) cs(Cookie) cs(Referer) cs-host sc-status sc-substatus sc-win32-status sc-bytes cs-bytes time-taken',
  u'2014-01-01 02:01:09 SAMPLEWEBSITE GET /blogposts/mvc4/step2.png X-ARR-LOG-ID=2ec4b8ad-3cf0-4442-93ab-837317ece6a1 80 - 1.54.23.196 Mozilla/5.0+(Windows+NT+6.3;+WOW64)+AppleWebKit/537.36+(KHTML,+like+Gecko)+Chrome/31.0.1650.63+Safari/537.36 - http://weblogs.asp.net/sample/archive/2007/12/09/asp-net-mvc-framework-part-4-handling-form-edit-and-post-scenarios.aspx www.sample.com 200 0 0 53175 871 46',
  u'2014-01-01 02:01:09 SAMPLEWEBSITE GET /blogposts/mvc4/step3.png X-ARR-LOG-ID=9eace870-2f49-4efd-b204-0d170da46b4a 80 - 1.54.23.196 Mozilla/5.0+(Windows+NT+6.3;+WOW64)+AppleWebKit/537.36+(KHTML,+like+Gecko)+Chrome/31.0.1650.63+Safari/537.36 - http://weblogs.asp.net/sample/archive/2007/12/09/asp-net-mvc-framework-part-4-handling-form-edit-and-post-scenarios.aspx www.sample.com 200 0 0 51237 871 32',
  u'2014-01-01 02:01:09 SAMPLEWEBSITE GET /blogposts/mvc4/step4.png X-ARR-LOG-ID=4bea5b3d-8ac9-46c9-9b8c-ec3e9500cbea 80 - 1.54.23.196 Mozilla/5.0+(Windows+NT+6.3;+WOW64)+AppleWebKit/537.36+(KHTML,+like+Gecko)+Chrome/31.0.1650.63+Safari/537.36 - http://weblogs.asp.net/sample/archive/2007/12/09/asp-net-mvc-framework-part-4-handling-form-edit-and-post-scenarios.aspx www.sample.com 200 0 0 72177 871 47']

三、使用自定义 Python 库分析日志数据

1. 在上面的输出中，前几行包括标头信息，其余的每一行均与此标头中描述的架构相匹配。 分析此类日志可能很复杂。 因此，可使用自定义 Python 库 (iislogparser.py)，它能使分析这类日志变得容易得多。 默认情况下，此库包含在 /HdiSamples/HdiSamples/WebsiteLogSampleData/iislogparser.py处 HDInsight 上的 Spark 群集中。但是，此库不在 PYTHONPATH 中，因此不能通过 import iislogparser 等导入语句来使用它。 要使用此库，必须将其分发给所有辅助角色节点。 运行以下代码段。

sc.addPyFile('wasb:///HdiSamples/HdiSamples/WebsiteLogSampleData/iislogparser.py')

2. 如果日志行是标题行，则 iislogparser 提供返回 None 的函数 parse_log_line，并且在遇到日志行时返回 LogLine 类的实例。 使用 LogLine 类从 RDD 中仅提取日志行：

def parse_line(l):
    import iislogparser
    return iislogparser.parse_log_line(l)
logLines = logs.map(parse_line).filter(lambda p: p is not None).cache()

3. 检索一些提取的日志行，以验证该步骤是否成功完成。

logLines.take(2)

输出应如下所示:

 2014-01-01 02:01:09 SAMPLEWEBSITE GET /blogposts/mvc4/step3.png X-ARR-LOG-ID=9eace870-2f49-4efd-b204-0d170da46b4a 80 - 1.54.23.196 Mozilla/5.0+(Windows+NT+6.3;+WOW64)+AppleWebKit/537.36+(KHTML,+like+Gecko)+Chrome/31.0.1650.63+Safari/537.36 - http://weblogs.asp.net/sample/archive/2007/12/09/asp-net-mvc-framework-part-4-handling-form-edit-and-post-scenarios.aspx www.sample.com 200 0 0 51237 871 32]

4. 反过来，LogLine 类具有一些有用的方法，如 is_error()，可返回日志条目是否具有错误代码。 使用此类计算提取日志行中的错误数，然后将所有错误记录到另一个文件中。

errors = logLines.filter(lambda p: p.is_error())
numLines = logLines.count()
numErrors = errors.count()
print 'There are', numErrors, 'errors and', numLines, 'log entries'
errors.map(lambda p: str(p)).saveAsTextFile('wasb:///HdiSamples/HdiSamples/WebsiteLogSampleData/SampleLog/909f2b-2.log')

应该看到如下输出：

There are 30 errors and 646 log entries

5. 还可使用 Matplotlib 构造数据的可视化效果。 例如，如果要找出请求长时间运行的原因，可能需要查找平均执行时间最长的文件。 下面的代码片段检索执行请求花费时间最长的前 25 个资源。

def avgTimeTakenByKey(rdd):
    return rdd.combineByKey(lambda line: (line.time_taken, 1),
                            lambda x, line: (x[0] + line.time_taken, x[1] + 1),
                            lambda x, y: (x[0] + y[0], x[1] + y[1]))
              .map(lambda x: (x[0], float(x[1][0]) / float(x[1][1])))

avgTimeTakenByKey(logLines.map(lambda p: (p.cs_uri_stem, p))).top(25, lambda x: x[1])

应该看到如下输出：

[(u'/blogposts/mvc4/step13.png', 197.5),
 (u'/blogposts/mvc2/step10.jpg', 179.5),
 (u'/blogposts/extractusercontrol/step5.png', 170.0),
 (u'/blogposts/mvc4/step8.png', 159.0),
 (u'/blogposts/mvcrouting/step22.jpg', 155.0),
 (u'/blogposts/mvcrouting/step3.jpg', 152.0),
 (u'/blogposts/linqsproc1/step16.jpg', 138.75),
 (u'/blogposts/linqsproc1/step26.jpg', 137.33333333333334),
 (u'/blogposts/vs2008javascript/step10.jpg', 127.0),
 (u'/blogposts/nested/step2.jpg', 126.0),
 (u'/blogposts/adminpack/step1.png', 124.0),
 (u'/BlogPosts/datalistpaging/step2.png', 118.0),
 (u'/blogposts/mvc4/step35.png', 117.0),
 (u'/blogposts/mvcrouting/step2.jpg', 116.5),
 (u'/blogposts/aboutme/basketball.jpg', 109.0),
 (u'/blogposts/anonymoustypes/step11.jpg', 109.0),
 (u'/blogposts/mvc4/step12.png', 106.0),
 (u'/blogposts/linq8/step0.jpg', 105.5),
 (u'/blogposts/mvc2/step18.jpg', 104.0),
 (u'/blogposts/mvc2/step11.jpg', 104.0),
 (u'/blogposts/mvcrouting/step1.jpg', 104.0),
 (u'/blogposts/extractusercontrol/step1.png', 103.0),
 (u'/blogposts/sqlvideos/sqlvideos.jpg', 102.0),
 (u'/blogposts/mvcrouting/step21.jpg', 101.0),
 (u'/blogposts/mvc4/step1.png', 98.0)]

6. 还可以图绘形式显示此信息。 创建绘图的第一步是创建一个临时表 AverageTime。 该表按照时间对日志进行分组，以查看在任何特定时间是否存在任何异常延迟峰值。

avgTimeTakenByMinute = avgTimeTakenByKey(logLines.map(lambda p: (p.datetime.minute, p))).sortByKey()
schema = StructType([StructField('Minutes', IntegerType(), True),
                     StructField('Time', FloatType(), True)])

avgTimeTakenByMinuteDF = spark.createDataFrame(avgTimeTakenByMinute, schema)
avgTimeTakenByMinuteDF.registerTempTable('AverageTime')

Warning:原文doc中的sqlContext.createDataFrame 需要改为spark.createDataFrame，不然会遇到'StructField' object has no attribute '_get_object_id'的报错

7. 接下来可以运行以下 SQL 查询以获取 AverageTime 表中的所有记录。

%%sql -o averagetime
SELECT * FROM AverageTime

后接 -o averagetime 的 %%sql magic 可确保查询输出本地保存在 Jupyter 服务器上（通常在群集的头结点）。 输出作为 Pandas 数据帧进行保存，指定名称为 averagetime

8. 现可使用 Matplotlib（用于构造数据效果可视化的库）创建绘图。 因为必须从本地保存的 averagetime 数据帧中创建绘图，所以代码片段必须以 %%local magic 开头。 这可确保代码在 Jupyter 服务器上本地运行。

%%local
%matplotlib inline
import matplotlib.pyplot as plt

plt.plot(averagetime['Minutes'], averagetime['Time'], marker='o', linestyle='--')
plt.xlabel('Time (min)')
plt.ylabel('Average time taken for request (ms)')
plt.grid(True)

9. 完成运行应用程序之后，应该要关闭笔记本以释放资源。 为此，请在笔记本的“文件”菜单中，单击“关闭并停止”。 这将会关闭 notebook。