Python中的探索性数据分析(功能式)
这里有一些技巧来处理日志文件提取。假设我们正在查看一些EnterpriseSplunk提取。我们可以用Splunk来探索数据。或者我们可以得到一个简单的提取并在Python中摆弄这些数据。
在Python中运行不同的实验似乎比试图在Splunk中进行这种探索性的操作更有效。主要是因为我们可以无所限制地对数据做任何事。我们可以在一个地方创建非常复杂的统计模型。
理论上，我们可以在Splunk中做很多的探索。它有各种报告和分析功能。
但是...
使用Splunk需要假设我们知道我们正在寻找什么。在很多情况下，我们不知道我们在寻找什么：我们正在探索。可能会有一些迹象表明，一些RESTfulAPI处理速度很慢，但还不止于此。我们如何继续？
第一步是获取CSV格式的原始数据。怎么办？
读取原始数据
我们将首先用一些附加函数来包装一个CSV.DictReader对象。
面向对象的纯粹主义者会反对这个策略。“为什么不扩展DictReader？”他们问。我没有一个很好的答案。我倾向于函数式编程和组件的正交性。对于一个纯粹的面向对象的方法，我们不得不使用更复杂的混合来实现这一点。
我们处理日志的一般框架是这样的。
withopen("somefile.csv")assource:
rdr=csv.DictReader(source)
这使我们可以读取CSV格式的Splunk提取物。我们可以迭代阅读器中的行。这是诀窍＃1。这不是非常棘手，但我喜欢它。
withopen("somefile.csv")assource:
rdr=csv.DictReader(source)
forrowinrdr:
print("{host}{ResponseTime}{source}{Service}".format_map(row))
我们可以-在一定程度上-以有用的格式报告原始数据。如果我们想粉饰一下输出，我们可以改变格式字符串。那就可能是“{主机：30s}{回复时间：8s}{来源：s}”或类似的东西。
过滤
常见的情况是我们提取了太多，但其实只需要看一个子集。我们可以更改Splunk过滤器，但是，在完成我们的探索之前，过量使用过滤器令人讨厌。在Python中过滤要容易得多。一旦我们了解到需要什么，就可以在Splunk中完成。
withopen("somefile.csv")assource:
rdr=csv.DictReader(source)
rdr_perf_log=(rowforrowinrdrifrow['source']=='perf_log')
forrowinrdr_perf_log:
print("{host}{ResponseTime}{Service}".format_map(row))
我们已经加入了一个生成器表达式来过滤源行，能够处理一个有意义的子集。
投影
在某些情况下，我们会添加额外的源数据列，这些列我们并不想使用。所以将通过对每一行进行投影来消除这些数据。
原则上，Splunk从不产生空列。但是，RESTfulAPI日志可能会导致数据集中包含大量列标题，这些列标题是基于请求URI一部分的代理键。这些列将包含来自使用该代理键的一个请求的一行数据。对于其他行，在这一列中没有任何用处。所以要删除这些空列。
我们也可以用一个生成器表达式来做到这一点，但是它会变得有点长。生成器函数更容易阅读.
defproject(reader):
forrowinreader:
yield{k:vfork,vinrow.items()ifv}
我们已经从原始阅读器中的一部分项目构建了一个新的行字典。我们可以使用它来包装我们的过滤器的输出。
withopen("somefile.csv")assource:
rdr=csv.DictReader(source)
rdr_perf_log=(rowforrowinrdrifrow['source']=='perf_log')
forrowinproject(rdr_perf_log):
print("{host}{ResponseTime}{Service}".format_map(row))
这将减少在for语句内部可见的未使用的列。
符号更改
row['source']符号会变得比较笨重。使用types.SimpleNamespace比用字典更好。这使得我们可以使用row.source。
这是一个很酷的技巧来创造更有用的东西。
rdr_ns=(types.SimpleNamespace(**row)forrowinreader)
我们可以将其折叠成这样的步骤序列。
withopen("somefile.csv")assource:
rdr=csv.DictReader(source)
rdr_perf_log=(rowforrowinrdrifrow['source']=='perf_log')
rdr_proj=project(rdr_perf_log)
rdr_ns=(types.SimpleNamespace(**row)forrowinrdr_proj)
forrowinrdr_ns:
print("{host}{ResponseTime}{Service}".format_map(vars(row)))
请注意我们对format_map（）方法的小改动。从SimpleNamespace的属性中，我们添加了vars（）函数来提取字典。
我们可以用其他函数把它写成一个函数来保留句法对称性。
defns_reader(reader): return(types.SimpleNamespace(**row)forrowinreader)
的确，我们可以把它写成一个像函数一样使用的lambda结构
ns_reader=lambdareader:(types.SimpleNamespace(**row)forrowinreader)
虽然ns_reader（）函数和ns_reader（）lambda的使用方式相同，但为lambda编写文档字符串和doctest单元测试稍微困难一些。出于这个原因，应该避免使用lambda结构。
我们可以使用map（lambdarow：types.SimpleNamespace（**row），reader）。有些人喜欢这个发生器表达式。
我们可以用一个适当的for语句和一个内部的yield语句，但是从一个小的东西里写大的语句似乎没有什么好处。
我们有很多选择，因为Python提供了如此多的函数式编程功能。虽然我们不会经常把Python视作一种功能性语言。但我们有多种方法来处理简单的映射。
映射：转换和派生数据
我们经常会有一个非常明显的数据转换列表。此外，我们将有一个衍生的数据项目越来越多的列表。衍生项目将是动态的，并基于我们正在测试的不同假设。每当我们有一个实验或问题，我们可能会改变派生的数据。
这些步骤中的每一个：过滤，投影，转换和派生都是map-reduce管道的“map”部分的阶段。我们可以创建一些较小的函数，并将其应用于map（）。因为我们正在更新一个有状态的对象，所以我们不能使用一般的map（）函数。如果我们想实现一个更纯粹的函数式编程风格，我们将使用一个不可变的namedtuple而不是一个可变的SimpleNamespace。
defconvert(reader): forrowinreader: row._time=datetime.datetime.strptime(row.Time,"%Y-%m-%dT%H:%M:%S.%F%Z") row.response_time=float(row.ResponseTime) yieldrow
在我们探索的过程中，我们将调整这个转换函数的主体。也许我们将从一些最小的转换和派生开始。我们将用一些“这些是正确的？”的问题来继续探索。当我们发现不工作时，我们会从中取出一些。
我们的整体处理过程如下所示：
withopen("somefile.csv")assource:
rdr=csv.DictReader(source)
rdr_perf_log=(rowforrowinrdrifrow['source']=='perf_log')
rdr_proj=project(rdr_perf_log)
rdr_ns=(types.SimpleNamespace(**row)forrowinrdr_proj)
rdr_converted=convert(rdr_ns)
forrowinrdr_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（）函数之前，我们会看看它们是否有用（甚至是正确的）。
减量
在减量方面，我们可以采取稍微不同的加工方式。我们需要重构我们之前的例子，并把它变成一个生成器函数。
defconverted_log(some_file): withopen(some_file)assource: rdr=csv.DictReader(source) rdr_perf_log=(rowforrowinrdrifrow['source']=='perf_log') rdr_proj=project(rdr_perf_log) rdr_ns=(types.SimpleNamespace(**row)forrowinrdr_proj) rdr_converted=convert(rdr_ns) forrowinrdr_converted: row.start_time=row._time-datetime.timedelta(seconds=row.response_time) row.service=some_mapping(row.Service) yieldrow
接着用一个yield代替了print（）。
这是重构的另一部分。
forrowinconverted_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)
forrowinconverted_log("somefile.csv"):
by_service[row.service]=row.response_time
forsvcinsorted(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”）中的行，我们可以通过一个小小的，易于修改的脚本探索很多处理选择。我们可以探索一些假设来确定为什么某些RESTfulAPI处理速度慢，而其他处理速度则很快。
总结
以上所述是小编给大家介绍的Python中的探索性数据分析(功能式)，希望对大家有所帮助，如果大家有任何疑问请给我留言，小编会及时回复大家的。在此也非常感谢大家对毛票票的支持！