Python函数式编程实现探索性数据分析（EDA）

处理日志文件提取时，有一些实用技巧。假设我们面对的是Enterprise Splunk导出的数据，既可以使用Splunk内置功能进行探索，也可以获取简单的提取文件并在Python环境中对数据进行深入操作。

在Python中运行各种实验，往往比试图在Splunk界面中进行探索性操作更为高效。主要原因在于Python允许对数据执行无限制的操作，例如在同一环境中构建非常复杂的统计模型。

理论上，Splunk具备多种报告和分析功能，能够支持大量的探索性工作。

然而，

使用Splunk通常隐含一个前提：我们知道自己在寻找什么。但在许多实际场景中，我们并不清楚目标所在，这正是探索性分析的意义。例如，可能观察到某些RESTful API处理速度较慢，但具体原因尚不明确。此时应如何继续？

第一步是获取原始数据的CSV格式文件。具体该如何操作？

读取原始数据

我们将首先定义一些辅助函数，用于包装csv.DictReader对象。

面向对象的纯粹主义者可能会反对这种策略，质问“为什么不直接扩展DictReader？”。对此我并无完美答案，但我更倾向于函数式编程以及组件之间的正交性。若采用纯粹的面向对象方法，则需要使用更复杂的混合模式来实现相同功能。

我们处理日志数据的一般框架如下所示。

with open("somefile.csv") as source:
rdr = csv.DictReader(source)

这使得我们能够读取Splunk导出的CSV格式数据，并迭代阅读器中的每一行。这是技巧之一，虽然不算非常复杂，但我认为这种写法值得推荐。

with open("somefile.csv") as source:
rdr = csv.DictReader(source)
for row in rdr:
print( "{host} {ResponseTime} {source} {Service}".format_map(row) )

我们可以在一定程度上以结构化格式报告原始数据。如果需要美化输出，可以调整格式字符串，例如使用“{host:30s} {ResponseTime:8s} {source:s}”等形式。

过滤

常见情况是我们提取的数据量过大，实际上只需关注其中一部分。虽然可以修改Splunk的过滤条件，但在探索阶段，过度使用过滤器会显得繁琐。在Python中进行过滤则容易得多，一旦明确需求，再回到Splunk中执行过滤也不迟。

with open("somefile.csv") as source:
rdr = csv.DictReader(source)
rdr_perf_log = (row for row in rdr if row['source'] == 'perf_log')
for row in rdr_perf_log:
print( "{host} {ResponseTime} {Service}".format_map(row) )

我们已加入生成器表达式以过滤源行，从而能够处理具有意义的子集数据。

投影

在某些情况下，源数据中会添加额外的列，而这些列并非我们所需。因此，我们需要对每一行进行投影，以剔除这些冗余数据。

原则上，Splunk不会生成空列。但是，RESTful API日志可能导致数据集中包含大量基于请求URI部分生成的代理键列。这些列仅在对应使用该代理键的请求行中有数据，其他行则为空。因此，需要删除这些包含空值的列。

虽然也可以使用生成器表达式实现，但代码可能会变得冗长。使用生成器函数则更易阅读。

def project(reader):
for row in reader:
yield {k:v for k,v in row.items() if v}

我们已从原始阅读器的一部分项目构建了一个新的行字典，并可用其包装过滤器的输出。

with open("somefile.csv") as source:
rdr = csv.DictReader(source)
rdr_perf_log = (row for row in rdr if row['source'] == 'perf_log')
for row in project(rdr_perf_log):
print( "{host} {ResponseTime} {Service}".format_map(row) )

这将减少在for语句内部可见的未使用列，使代码更清晰。

符号更改

使用row['source']符号会变得较为笨重。改用types.SimpleNamespace比直接使用字典更好，这使得我们可以通过row.source直接访问属性。

这是一个非常实用的技巧，能创造出更易于使用的数据结构。

rdr_ns= (types.SimpleNamespace(**row) forrowinreader)

我们可以将上述步骤折叠为如下序列。

with open("somefile.csv") as source:
rdr = csv.DictReader(source)
rdr_perf_log = (row for row in rdr if row['source'] == 'perf_log')
rdr_proj = project(rdr_perf_log)
rdr_ns = (types.SimpleNamespace(**row) for row in rdr_proj)
for row in rdr_ns:
print( "{host} {ResponseTime} {Service}".format_map(vars(row)) )

请注意我们对format_map（）方法所做的微小改动。从SimpleNamespace的属性中，我们添加了vars()函数来提取字典，以便格式化输出。

我们可以用其他函数将其封装，以保持语法对称性。

def ns_reader(reader):
return (types.SimpleNamespace(**row) for row in reader)

确实，我们可以将其写成一个类似函数的lambda结构。

ns_reader = lambda reader: (types.SimpleNamespace(**row) for row in reader)

虽然ns_reader()函数与ns_reader() lambda的使用方式相同，但为lambda编写文档字符串和doctest单元测试相对困难。因此，出于可维护性考虑，应尽量避免使用lambda结构。

我们可以使用map（lambda row：types.SimpleNamespace（** row），reader）。，有些人喜欢这种生成器表达式写法。

虽然可以用一个包含内部yield语句的普通for语句实现，但从简单事物中构建复杂语句似乎并无必要。

由于Python提供了丰富的函数式编程功能，我们拥有多种处理简单映射的方法。尽管我们并不常将Python视为纯粹的功能性语言，但处理简单映射时仍有多种选择。

映射：转换和派生数据

我们通常会遇到明显的数据转换列表，以及日益增加的衍生数据项目。衍生项目是动态的，基于我们正在测试的不同假设。每当进行实验或提出问题时，我们可能会改变衍生数据的定义。

这些步骤中的每一个：过滤、投影、转换和派生，都是Map-Reduce管道中“Map”部分的阶段。我们可以创建一些较小的函数，并将其应用于map()。由于我们正在更新一个有状态的对象，因此不能使用通用的map()函数。若希望实现更纯粹的函数式编程风格，应使用不可变的namedtuple替代可变的SimpleNamespace。

def convert(reader):
for row in reader:
row._time = datetime.datetime.strptime(row.Time, "%Y-%m-%dT%H:%M:%S.%F%Z")
row.response_time = float(row.ResponseTime)
yield row

在探索过程中，我们将不断调整这个转换函数的主体。也许我们从最简单的转换和派生开始，随后通过提出“这些是正确的吗？”等问题继续探索。当发现某些部分不工作时，我们会移除它们。

我们的整体处理流程如下所示：

with open("somefile.csv") as source:
rdr = csv.DictReader(source)
rdr_perf_log = (row for row in rdr if row['source'] == 'perf_log')
rdr_proj = project(rdr_perf_log)
rdr_ns = (types.SimpleNamespace(**row) for row in rdr_proj)
rdr_converted = convert(rdr_ns)
for row in rdr_converted:
row.start_time = row._time - datetime.timedelta(seconds=row.response_time)
row.service = some_mapping(row.Service)
print( "{host:30s} {start_time:%H:%M:%S} {response_time:6.3f} {service}".format_map(vars(row)) )

请注意语句主体的变化。convert()函数生成我们确定的值。我们在for循环中添加了几个额外变量，目前尚不能100%确定其有效性。在更新convert()函数之前，我们会先观察这些变量是否有用（或正确）。

归约（Reduce）

在归约方面，我们可以采取稍有不同的处理方式。我们需要重构之前的示例，将其变为一个生成器函数。

def converted_log(some_file):
with open(some_file) as source:
rdr = csv.DictReader(source)
rdr_perf_log = (row for row in rdr if row['source'] == 'perf_log')
rdr_proj = project(rdr_perf_log)
rdr_ns = (types.SimpleNamespace(**row) for row in rdr_proj)
rdr_converted = convert(rdr_ns)
for row in rdr_converted:
row.start_time = row._time - datetime.timedelta(seconds=row.response_time)
row.service = some_mapping(row.Service)
yield row

接着，用yield语句替代print()。

这是重构的另一部分。

for row in converted_log("somefile.csv"):
print( "{host:30s} {start_time:%H:%M:%S} {response_time:6.3f} {service}".format_map(vars(row)) )

理想情况下，我们的所有编程都应如此。我们使用生成器函数来生成数据，而数据的最终显示保持完全分离。这使我们能够更自由地重构和更改处理逻辑。

现在我们可以执行一些操作，例如将行收集到Counter()对象中，或计算某些统计信息。我们可以使用defaultdict(list)按服务对行进行分组。

by_service= defaultdict(list)
for row in converted_log("somefile.csv"):
by_service[row.service] = row.response_time
for svc in sorted(by_service):
m = statistics.mean( by_service[svc] )
print( "{svc:15s} {m:.2f}".format_map(vars()) )

我们决定在此处创建具体的列表对象。可以使用itertools按服务分组响应时间。这看起来符合函数式编程规范，但这种实施在Pythonic函数式编程形式中揭示了一些限制：要么必须对数据进行排序（从而创建列表对象），要么在分组数据时创建列表。为了进行几种不同的统计计算，通过创建具体列表来分组数据通常更为简便。

我们现在正在执行两项任务，而不仅仅是打印行对象。

创建一些局部变量，如svc和m。我们可以轻松添加变化或其他度量指标。

使用不带参数的vars()函数，它会从局部变量中创建一个字典。

这种不带参数的vars()行为类似于locals()，是一个便捷的技巧。它允许我们简单地创建所需的任何局部变量，并将它们包含在格式化输出中。我们可以侵入我们认为可能相关的各种统计方法中。

既然我们的基本处理循环是针对converted_log（“somefile.csv”）中的行，我们可以通过一个小型且易于修改的脚本探索多种处理选择。我们可以探索各种假设，以确定为何某些RESTful API处理速度慢，而其他则较快。

总结

以上所述是小编给大家介绍的Python中的探索性数据分析(功能式)，希望对大家有所帮助，如果大家有任何疑问请给我留言，小编会及时回复大家的。在此也非常感谢大家对本站的支持！