如何使用Python和Apache Spark分析日志数据-白红宇

如何使用Python和Apache Spark分析日志数据

阅读量：2532 次

发布时间：2019-05-11

本文共 18845 字，大约阅读时间需要 62 分钟。

在本系列的中，我们首先使用和将示例Web日志处理和整理为适合分析的格式，这是考虑到当今大多数组织生成的大量日志数据的一项至关重要的技术。我们设置了环境变量，依赖项，并加载了使用DataFrame和正则表达式所需的库，当然还加载了示例日志数据。然后，我们将日志数据整理为干净，结构化且有意义的格式。在第二部分中，我们重点分析数据。

网络日志中的数据分析

现在，我们有了一个包含经过分析和清理的日志文件作为数据帧的，我们可以执行一些有趣的探索性数据分析（EDA），以尝试获取一些有趣的见解！

内容大小统计

让我们计算一些有关Web服务器返回的内容大小的统计信息。特别是，我们想知道平均，最小和最大内容大小。

我们通过调用计算这些统计.describe()在content_size列logs_df 。 .describe()函数以以下格式返回给定列的count ， mean ， stddev ， min和max ：

content_size_summary_df      
     = logs_df.      
     describe      
     (      
     [      
     'content_size'      
     ]      
     )      
     content_size_summary_df.      
     toPandas      
     (      
     )

Stastical analysis regarding the size of content your web server returns.

^{有关Web服务器返回的内容大小的基本分析。}

或者，我们可以使用直接计算这些统计信息。 pyspark.sql.functions模块具有许多有用的功能，请参阅以了解更多信息。

应用.agg()函数后，我们调用toPandas()提取结果并将其转换为 DataFrame，从而在上提供更好的格式：

from pyspark.      
     sql      
     import functions      
     as F     
          
          
     ( logs_df.      
     agg      
     ( F.      
     min      
     ( logs_df      
     [      
     'content_size'      
     ]      
     ) .      
     alias      
     (      
     'min_content_size'      
     )      
     ,      
                  F.      
     max      
     ( logs_df      
     [      
     'content_size'      
     ]      
     ) .      
     alias      
     (      
     'max_content_size'      
     )      
     ,      
                  F.      
     mean      
     ( logs_df      
     [      
     'content_size'      
     ]      
     ) .      
     alias      
     (      
     'mean_content_size'      
     )      
     ,      
                  F.      
     stddev      
     ( logs_df      
     [      
     'content_size'      
     ]      
     ) .      
     alias      
     (      
     'std_content_size'      
     )      
     ,      
                  F.      
     count      
     ( logs_df      
     [      
     'content_size'      
     ]      
     ) .      
     alias      
     (      
     'count_content_size'      
     )      
     )      
             .      
     toPandas      
     (      
     )      
     )

^{相同的数据重新格式化为熊猫数据框。}

当我们验证结果时，我们看到它们与预期的相同。

HTTP状态码分析

接下来，让我们查看日志的状态码值，以查看出现哪些状态码值以及显示多少次。我们再次从logs_df开始，按status列分组，应用.count()聚合函数，然后按status列排序：

status_freq_df      
     =      
     ( logs_df     
                          .      
     groupBy      
     (      
     'status'      
     )      
                          .      
     count      
     (      
     )      
                          .      
     sort      
     (      
     'status'      
     )      
                          .      
     cache      
     (      
     )      
     )      
          
     print      
     (      
     'Total distinct HTTP Status Codes:'      
     , status_freq_df.      
     count      
     (      
     )      
     )

Total Distinct HTTP Status Codes: 8

让我们以频率表的形式查看每个状态码的出现：

status_freq_pd_df      
     =      
     ( status_freq_df     
                              .      
     toPandas      
     (      
     )      
                              .      
     sort_values      
     ( by      
     =      
     [      
     'count'      
     ]      
     ,      
                                           ascending      
     =      
     False      
     )      
     )      
     status_freq_pd_df

How many times each status code appears in your log.

^{每个状态代码出现在日志中的次数。}

看起来，最常见的状态码是200（确定），这是大多数情况下正常工作的一个好兆头。让我们将其可视化：

import matplotlib.      
     pyplot      
     as plt     
          
     import seaborn      
     as sns     
          
     import numpy      
     as np     
     %matplotlib inline     
          
     sns.      
     catplot      
     ( x      
     =      
     'status'      
     , y      
     =      
     'count'      
     , data      
     = status_freq_pd_df      
     ,      
                 kind      
     =      
     'bar'      
     , order      
     = status_freq_pd_df      
     [      
     'status'      
     ]      
     )

HTTP status code occurrences in a bar chart.

^{条形图中出现的HTTP状态代码。}

还不错但是由于数据的巨大偏差，几乎看不到几个状态代码。让我们进行日志转换，看看情况是否有所改善。通常，对数转换可帮助我们将高度偏斜的数据转换为近似正态分布，以便我们可以以更易理解的方式可视化数据分布：

log_freq_df      
     = status_freq_df.      
     withColumn      
     (      
     'log(count)'      
     ,      
                                             F.      
     log      
     ( status_freq_df      
     [      
     'count'      
     ]      
     )      
     )      
     log_freq_df.      
     show      
     (      
     )

Error code frequency as a log transform.

^{错误代码频率作为对数变换。}

log_freq_pd_df      
     =      
     ( log_freq_df     
                         .      
     toPandas      
     (      
     )      
                         .      
     sort_values      
     ( by      
     =      
     [      
     'log(count)'      
     ]      
     ,      
                                      ascending      
     =      
     False      
     )      
     )      
     sns.      
     catplot      
     ( x      
     =      
     'status'      
     , y      
     =      
     'log(count)'      
     , data      
     = log_freq_pd_df      
     ,      
                 kind      
     =      
     'bar'      
     , order      
     = status_freq_pd_df      
     [      
     'status'      
     ]      
     )

HTTP status code frequency bar chart, after a log transform.

^{对数转换后的HTTP状态码频率条形图。}

该图表看起来肯定好多了，而且偏斜也更少了，这使我们对状态码的分布有了更好的了解！

分析频繁的主机

通过获取每个主机的访问总数，按访问次数排序并仅显示前10个最频繁的主机，来查看频繁访问服务器的主机：

host_sum_df      
     =      
     ( logs_df     
                    .      
     groupBy      
     (      
     'host'      
     )      
                    .      
     count      
     (      
     )      
                    .      
     sort      
     (      
     'count'      
     , ascending      
     =      
     False      
     ) .      
     limit      
     (      
     10      
     )      
     )      
          
     host_sum_df.      
     show      
     ( truncate      
     =      
     False      
     )

Hosts that frequently access the server sorted by number of accesses.

^{频繁访问服务器的主机按访问次数排序。}

该表看起来不错，但是让我们更仔细地检查第9行中的空白记录：

host_sum_pd_df      
     = host_sum_df.      
     toPandas      
     (      
     )      
     host_sum_pd_df.      
     iloc      
     [      
     8      
     ]      
     [ ‘host’      
     ]

''

看起来顶级主机名之一是一个空字符串。这个例子教了我们一个宝贵的教训：不仅在数据处理时检查空值，还检查空字符串。

显示最常见的前20个端点

现在，让我们可视化日志中端点点击次数。要执行此任务，请从logs_df开始，然后按端点列进行分组，按计数进行聚合，然后按降序排序（如上例所示）：

paths_df      
     =      
     ( logs_df     
                 .      
     groupBy      
     (      
     'endpoint'      
     )      
                 .      
     count      
     (      
     )      
                 .      
     sort      
     (      
     'count'      
     , ascending      
     =      
     False      
     ) .      
     limit      
     (      
     20      
     )      
     )      
          
     paths_pd_df      
     = paths_df.      
     toPandas      
     (      
     )      
     paths_pd_df

A table showing the number of hits to each endpoint URI in descending order.

^{该表以降序显示每个端点URI的命中数。}

毫不奇怪，访问最多的资产是GIF，主页和一些CGI脚本。

显示前10个错误端点

请求的前10个没有返回码200（HTTP状态正常）的端点是什么？为了找出答案，我们创建了一个排序列表，其中包含端点以及使用非200返回码访问它们的次数，然后显示前10个：

not200_df      
     =      
     ( logs_df     
                    .      
     filter      
     ( logs_df      
     [      
     'status'      
     ]      
     !=      
     200      
     )      
     )      
          
     error_endpoints_freq_df      
     =      
     ( not200_df     
                                    .      
     groupBy      
     (      
     'endpoint'      
     )      
                                    .      
     count      
     (      
     )      
                                    .      
     sort      
     (      
     'count'      
     , ascending      
     =      
     False      
     )      
                                    .      
     limit      
     (      
     10      
     )      
                                    
     )      
                                    
     error_endpoints_freq_df.      
     show      
     ( truncate      
     =      
     False      
     )

A table displaying the top ten error endpoints and their frequency.

^{表格显示前十个错误端点及其频率。}

看起来GIF（动画/静态图像）加载最多。为什么会这样呢？考虑到这些日志来自1995年，并且考虑到我们当时的互联网速度，我并不感到惊讶！

唯一主机总数

在这两个月中，有多少独特的主机访问过NASA网站？我们可以通过一些转换找到答案：

unique_host_count      
     =      
     ( logs_df     
                          .      
     select      
     (      
     'host'      
     )      
                          .      
     distinct      
     (      
     )      
                          .      
     count      
     (      
     )      
     )      
     unique_host_count

137933

每日唯一主机数

作为一个高级示例，让我们看一下如何每天确定唯一主机的数量。在这里，我们想要一个DataFrame，其中包含每月的某天以及该天的唯一主机的相关数量，并按该月份的递增天数进行排序。

考虑执行此任务所需执行的步骤。由于这些日志仅覆盖一个月，因此至少可以忽略月份问题。对于跨越多个月的数据，我们在进行必要的汇总时需要同时考虑月份和日期。您可能要使用pyspark.sql.functions模块的函数（在本教程开始时我们已经导入为F ) 。

从host_day_df开始，这是一个具有两列的DataFrame：

The columns in the host_day_df dataframe.

^{host_day_df数据框中的列。}

有一个在这个数据帧中的每一行一行logs_df 。本质上，我们只是在变换每一行。例如，对于此行：

unicomp6. unicomp . net - - [ 01 /Aug/ 1995 : 00 : 35 : 41 - 0400 ] "GET /shuttle/missions/sts-73/news HTTP/1.0" 302 -

您的host_day_df应该具有unicomp6.unicomp.net 1

host_day_df      
     = logs_df.      
     select      
     ( logs_df.      
     host      
     ,      
                                  F.      
     dayofmonth      
     (      
     'time'      
     ) .      
     alias      
     (      
     'day'      
     )      
     )      
     host_day_df.      
     show      
     (      
     5      
     , truncate      
     =      
     False      
     )

The top five hosts making requests on the first day.

^{前五名提出请求的主机。}

接下来是host_day_distinct_df ，这是一个与host_day_df相同的列的host_day_df ，但是删除了重复的(day, host)行：

host_day_df      
     = logs_df.      
     select      
     ( logs_df.      
     host      
     ,      
                                  F.      
     dayofmonth      
     (      
     'time'      
     ) .      
     alias      
     (      
     'day'      
     )      
     )      
     host_day_df.      
     show      
     (      
     5      
     , truncate      
     =      
     False      
     )

host_day_distinct_df gives the same output as host_day_df, but with duplicate rows removed.

^{host_day_distinct_df提供与host_day_df相同的输出，但删除了重复的行。}

另一个选项是daily_unique_hosts_df ，它是一个具有两列的DataFrame，与先前的DataFrame不同：

^{daily_unique_hosts_df显示的列。}

def_mr      
     = pd.      
     get_option      
     (      
     'max_rows'      
     )      
     pd.      
     set_option      
     (      
     'max_rows'      
     ,      
     10      
     )      
          
     daily_hosts_df      
     =      
     ( host_day_distinct_df     
                          .      
     groupBy      
     (      
     'day'      
     )      
                          .      
     count      
     (      
     )      
                          .      
     sort      
     (      
     "day"      
     )      
     )      
          
     daily_hosts_df      
     = daily_hosts_df.      
     toPandas      
     (      
     )      
     daily_hosts_df

daily_unique_hosts_df shows the day of the month and the number of unique hosts making requests that day.

^{daily_unique_hosts_df显示每月的某天以及当天发出请求的唯一主机的数量。}

这个结果为我们提供了一个不错的DataFrame，它显示了每天唯一主机的总数。让我们将其可视化：

c      
     = sns.      
     catplot      
     ( x      
     =      
     'day'      
     , y      
     =      
     'count'      
     ,      
                     data      
     = daily_hosts_df      
     ,      
                     kind      
     =      
     'point'      
     , height      
     =      
     5      
     ,      
                     aspect      
     =      
     1.5      
     )

Unique hosts per day charted using daily_unique_hosts_df.

^{每天的唯一主机图表为daily_unique_hosts_df。}

每个主机的平均每日请求数

在前面的示例中，我们研究了一种确定每天日志中唯一主机数量的方法。现在，让我们找到每个主机每天向NASA网站发出的平均请求数量。在这里，我们想要一个DataFrame，它按照每月增加的天数进行排序，其中包括每月的某天以及每个主机当日的平均请求数：

daily_hosts_df      
     =      
     ( host_day_distinct_df     
                          .      
     groupBy      
     (      
     'day'      
     )      
                          .      
     count      
     (      
     )      
                          .      
     select      
     ( col      
     (      
     "day"      
     )      
     ,      
                                           col      
     (      
     "count"      
     ) .      
     alias      
     (      
     "total_hosts"      
     )      
     )      
     )      
          
     total_daily_reqests_df      
     =      
     ( logs_df     
                                   .      
     select      
     ( F.      
     dayofmonth      
     (      
     "time"      
     )      
                                               .      
     alias      
     (      
     "day"      
     )      
     )      
                                   .      
     groupBy      
     (      
     "day"      
     )      
                                   .      
     count      
     (      
     )      
                                   .      
     select      
     ( col      
     (      
     "day"      
     )      
     ,      
                                           col      
     (      
     "count"      
     ) .      
     alias      
     (      
     "total_reqs"      
     )      
     )      
     )      
          
     avg_daily_reqests_per_host_df      
     = total_daily_reqests_df.      
     join      
     ( daily_hosts_df      
     ,      
     'day'      
     )      
     avg_daily_reqests_per_host_df      
     =      
     ( avg_daily_reqests_per_host_df     
                                         .      
     withColumn      
     (      
     'avg_reqs'      
     , col      
     (      
     'total_reqs'      
     ) / col      
     (      
     'total_hosts'      
     )      
     )      
                                         .      
     sort      
     (      
     "day"      
     )      
     )      
     avg_daily_reqests_per_host_df      
     = avg_daily_reqests_per_host_df.      
     toPandas      
     (      
     )      
     avg_daily_reqests_per_host_df

The average number of daily requests per host via avg_daily_reqests_per_host_df.

^{通过avg_daily_reqests_per_host_df每个主机的平均每日请求数。}

现在我们可以可视化每个主机的平均每日请求：

c      
     = sns.      
     catplot      
     ( x      
     =      
     'day'      
     , y      
     =      
     'avg_reqs'      
     ,      
                     data      
     = avg_daily_reqests_per_host_df      
     ,      
                     kind      
     =      
     'point'      
     , height      
     =      
     5      
     , aspect      
     =      
     1.5      
     )

The average number of daily requests per host charted.

^{列出每个主机的平均每日请求数。}

看起来第13天获得了每个主机的最大请求数。

计数404响应码

创建一个仅包含状态代码为404（未找到）的日志记录的DataFrame。我们确保将not_found_df DataFrame cache() ，因为我们将在此处的其余示例中使用它。您认为日志中有404条记录？

not_found_df      
     = logs_df.      
     filter      
     ( logs_df      
     [      
     "status"      
     ]      
     ==      
     404      
     ) .      
     cache      
     (      
     )      
          
     print      
     (      
     (      
     'Total 404 responses: {}'      
     ) .      
     format      
     ( not_found_df.      
     count      
     (      
     )      
     )      
     )

Total 404 responses: 20899

列出前二十个404响应代码端点

使用我们之前缓存的DataFrame（仅包含带有404响应代码的日志记录），我们现在将打印出产生404错误最多的前二十个端点的列表。请记住，每当您生成最高端点时，它们都应按排序顺序排列：

endpoints_404_count_df      
     =      
     ( not_found_df     
                               .      
     groupBy      
     (      
     "endpoint"      
     )      
                               .      
     count      
     (      
     )      
                               .      
     sort      
     (      
     "count"      
     , ascending      
     =      
     False      
     )      
                               .      
     limit      
     (      
     20      
     )      
     )      
          
     endpoints_404_count_df.      
     show      
     ( truncate      
     =      
     False      
     )

The top 20 response code endpoints, sorted, thanks to endpoints_404_count_df.

^{多亏了endpoints_404_count_df ，排名靠前的20个响应代码端点。}

列出前二十位的404响应代码主机

使用我们之前缓存的DataFrame，其中仅包含带有404响应代码的日志记录，现在我们可以打印出生成最多404错误的前二十个主机的列表。同样，请记住，顶级主机应按排序顺序：

hosts_404_count_df      
     =      
     ( not_found_df     
                               .      
     groupBy      
     (      
     "host"      
     )      
                               .      
     count      
     (      
     )      
                               .      
     sort      
     (      
     "count"      
     , ascending      
     =      
     False      
     )      
                               .      
     limit      
     (      
     20      
     )      
     )      
          
     hosts_404_count_df.      
     show      
     ( truncate      
     =      
     False      
     )

The top twenty 404 response code hosts via hosts_404_count_df.

^{前二十个404响应代码通过hosts_404_count_df托管。}

此输出为我们提供了一个好主意，即主机最终为NASA网页生成最多404错误。

每天可视化404个错误

现在让我们暂时（按时间）浏览404记录。与显示每日唯一主机数的示例类似，我们errors_by_date_sorted_df天细分404请求， errors_by_date_sorted_df逐日存储每日计数：

errors_by_date_sorted_df      
     =      
     ( not_found_df     
                                     .      
     groupBy      
     ( F.      
     dayofmonth      
     (      
     'time'      
     ) .      
     alias      
     (      
     'day'      
     )      
     )      
                                     .      
     count      
     (      
     )      
                                     .      
     sort      
     (      
     "day"      
     )      
     )      
          
     errors_by_date_sorted_pd_df      
     = errors_by_date_sorted_df.      
     toPandas      
     (      
     )      
     errors_by_date_sorted_pd_df

404 errors per day via errors_by_date_sorted_df.

^{每天通过errors_by_date_sorted_df发生404错误。}

现在，让我们现在可视化每天总共有404个错误：

c      
     = sns.      
     catplot      
     ( x      
     =      
     'day'      
     , y      
     =      
     'count'      
     ,      
                     data      
     = errors_by_date_sorted_pd_df      
     ,      
                     kind      
     =      
     'point'      
     , height      
     =      
     5      
     , aspect      
     =      
     1.5      
     )

^{每天总计404错误。}

前三天出现404错误

根据之前的图表，一个月中前三天的404错误最多？知道这一点可以帮助我们诊断和深入研究这些特定的日子，以找出可能出了问题的地方（服务器问题，DNS问题，拒绝服务，延迟问题，维护等）。我们可以利用先前创建的errors_by_date_sorted_df数据框来回答此问题：

( errors_by_date_sorted_df     
         .      
     sort      
     (      
     "count"      
     , ascending      
     =      
     False      
     )      
         .      
     show      
     (      
     3      
     )      
     )

The top 3 days of 404 errors via errors_by_date_sorted_df.

^{通过errors_by_date_sorted_df发出的404错误的前3天。}

可视化每小时404错误

使用我们先前缓存的DataFrame not_found_df ，我们现在可以按一天中的小时按升序进行分组和排序。我们将使用此过程创建一个DataFrame，其中包含一天中每个小时（午夜从0开始）的HTTP请求的404响应总数。然后，我们将从DataFrame建立可视化。

hourly_avg_errors_sorted_df      
     =      
     ( not_found_df     
                                        .      
     groupBy      
     ( F.      
     hour      
     (      
     'time'      
     )      
                                                  .      
     alias      
     (      
     'hour'      
     )      
     )      
                                        .      
     count      
     (      
     )      
                                        .      
     sort      
     (      
     'hour'      
     )      
     )      
     hourly_avg_errors_sorted_pd_df      
     = hourly_avg_errors_sorted_df.      
     toPandas      
     (      
     )      
          
     c      
     = sns.      
     catplot      
     ( x      
     =      
     'hour'      
     , y      
     =      
     'count'      
     ,      
                     data      
     = hourly_avg_errors_sorted_pd_df      
     ,      
                     kind      
     =      
     'bar'      
     , height      
     =      
     5      
     , aspect      
     =      
     1.5      
     )

Total 404 errors per hour in a bar chart.

^{条形图中每小时总计404错误。}

看起来404错误最多发生在下午，而最少发生在清晨。现在，我们可以将熊猫显示的最大行数重置为默认值，因为我们之前已对其进行了更改，以显示有限的行数。

pd. set_option ( ‘max_rows’ , def_mr )

结论

我们在Log Analytics的一个非常常见但必不可少的案例研究中，采用了动手的方法来大规模处理数据整理，解析，分析和可视化。尽管从大小或卷的角度来看，我们在这里处理的数据可能不是“大数据”，但这些技术和方法具有足够的通用性，可以扩展到更大的数据量。我希望本练习为您提供有关如何利用Apache Spark等开源框架处理自己的结构化和半结构化数据的想法！

您可以在中本文随附的所有代码和分析。另外，您可以在此找到分步方法。

有兴趣了解Spark SQL和DataFrames吗？在查看我的！

如果您有任何反馈或疑问，可以在这里发表评论或通过与我联系。

这篇文章最初出现在Medium的频道上，经许可重新发布。