摘要：在中運(yùn)行不同的實(shí)驗(yàn)似乎比試圖在中進(jìn)行這種探索性的操作更有效。理論上，我們可以在中做很多的探索。我們?nèi)绾卫^續(xù)第一步是獲取格式的原始數(shù)據(jù)。這些列將包含來自使用該代理鍵的一個(gè)請(qǐng)求的一行數(shù)據(jù)。這是重構(gòu)的另一部分。數(shù)據(jù)的最終顯示保持完全分離。

歡迎大家前往騰訊云+社區(qū)，獲取更多騰訊海量技術(shù)實(shí)踐干貨哦~

這里有一些技巧來處理日志文件提取。假設(shè)我們正在查看一些Enterprise Splunk提取。我們可以用Splunk來探索數(shù)據(jù)?；蛘呶覀兛梢缘玫揭粋€(gè)簡單的提取并在Python中擺弄這些數(shù)據(jù)。

在Python中運(yùn)行不同的實(shí)驗(yàn)似乎比試圖在Splunk中進(jìn)行這種探索性的操作更有效。主要是因?yàn)槲覀兛梢詿o所限制地對(duì)數(shù)據(jù)做任何事。我們可以在一個(gè)地方創(chuàng)建非常復(fù)雜的統(tǒng)計(jì)模型。

理論上，我們可以在Splunk中做很多的探索。它有各種報(bào)告和分析功能。

但是，使用Splunk需要假設(shè)我們知道我們正在尋找什么。在很多情況下，我們不知道我們?cè)趯ふ沂裁矗何覀冋谔剿??？赡軙?huì)有一些跡象表明，一些RESTful API處理速度很慢，但還不止于此。我們?nèi)绾卫^續(xù)？

第一步是獲取CSV格式的原始數(shù)據(jù)。怎么辦？

我們將首先用一些附加函數(shù)來包裝一個(gè)CSV.DictReader對(duì)象。

面向?qū)ο蟮募兇庵髁x者會(huì)反對(duì)這個(gè)策略。 “為什么不擴(kuò)展DictReader？”他們問。我沒有一個(gè)很好的答案。我傾向于函數(shù)式編程和組件的正交性。對(duì)于一個(gè)純粹的面向?qū)ο蟮姆椒?，我們不得不使用更?fù)雜的混合來實(shí)現(xiàn)這一點(diǎn)。

我們處理日志的一般框架是這樣的。

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

這使我們可以讀取CSV格式的Splunk提取物。我們可以迭代閱讀器中的行。這是訣竅＃1。這不是非常棘手，但我喜歡它。

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

我們可以 - 在一定程度上 - 以有用的格式報(bào)告原始數(shù)據(jù)。如果我們想粉飾一下輸出，我們可以改變格式字符串。那就可能是“{主機(jī)：30s} {回復(fù)時(shí)間：8s} {來源：s}”或類似的東西。

常見的情況是我們提取了太多，但其實(shí)只需要看一個(gè)子集。我們可以更改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) )

我們已經(jīng)加入了一個(gè)生成器表達(dá)式來過濾源行，能夠處理一個(gè)有意義的子集。

在某些情況下，我們會(huì)添加額外的源數(shù)據(jù)列，這些列我們并不想使用。所以將通過對(duì)每一行進(jìn)行投影來消除這些數(shù)據(jù)。

原則上，Splunk從不產(chǎn)生空列。但是，RESTful API日志可能會(huì)導(dǎo)致數(shù)據(jù)集中包含大量列標(biāo)題，這些列標(biāo)題是基于請(qǐng)求URI一部分的代理鍵。這些列將包含來自使用該代理鍵的一個(gè)請(qǐng)求的一行數(shù)據(jù)。對(duì)于其他行，在這一列中沒有任何用處。所以要?jiǎng)h除這些空列。

我們也可以用一個(gè)生成器表達(dá)式來做到這一點(diǎn)，但是它會(huì)變得有點(diǎn)長。生成器函數(shù)更容易閱讀。

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

我們已經(jīng)從原始閱讀器中的一部分項(xiàng)目構(gòu)建了一個(gè)新的行字典。我們可以使用它來包裝我們的過濾器的輸出。

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語句內(nèi)部可見的未使用的列。

row["source"]符號(hào)會(huì)變得比較笨重。使用types.SimpleNamespace比用字典更好。這使得我們可以使用row.source。

這是一個(gè)很酷的技巧來創(chuàng)造更有用的東西。

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)) )

請(qǐng)注意我們對(duì)format_map（）方法的小改動(dòng)。從SimpleNamespace的屬性中，我們添加了vars（）函數(shù)來提取字典 。

我們可以用其他函數(shù)把它寫成一個(gè)函數(shù)來保留句法對(duì)稱性。

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

的確，我們可以把它寫成一個(gè)像函數(shù)一樣使用的lambda結(jié)構(gòu)

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

雖然ns_reader（）函數(shù)和ns_reader（）lambda的使用方式相同，但為lambda編寫文檔字符串和doctest單元測(cè)試稍微困難一些。出于這個(gè)原因，應(yīng)該避免使用lambda結(jié)構(gòu)。

我們可以使用map（lambda row：types.SimpleNamespace（** row），reader）。有些人喜歡這個(gè)發(fā)生器表達(dá)式。

我們可以用一個(gè)適當(dāng)?shù)膄or語句和一個(gè)內(nèi)部的yield語句，但是從一個(gè)小的東西里寫大的語句似乎沒有什么好處。

我們有很多選擇，因?yàn)镻ython提供了如此多的函數(shù)式編程功能。雖然我們不會(huì)經(jīng)常把Python視作一種功能性語言。但我們有多種方法來處理簡單的映射。

我們經(jīng)常會(huì)有一個(gè)非常明顯的數(shù)據(jù)轉(zhuǎn)換列表。此外，我們將有一個(gè)衍生的數(shù)據(jù)項(xiàng)目越來越多的列表。衍生項(xiàng)目將是動(dòng)態(tài)的，并基于我們正在測(cè)試的不同假設(shè)。每當(dāng)我們有一個(gè)實(shí)驗(yàn)或問題，我們可能會(huì)改變派生的數(shù)據(jù)。

這些步驟中的每一個(gè)：過濾，投影，轉(zhuǎn)換和派生都是map-reduce管道的“map”部分的階段。我們可以創(chuàng)建一些較小的函數(shù)，并將其應(yīng)用于map（）。因?yàn)槲覀冋诟乱粋€(gè)有狀態(tài)的對(duì)象，所以我們不能使用一般的map（）函數(shù)。如果我們想實(shí)現(xiàn)一個(gè)更純粹的函數(shù)式編程風(fēng)格，我們將使用一個(gè)不可變的namedtuple而不是一個(gè)可變的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

在我們探索的過程中，我們將調(diào)整這個(gè)轉(zhuǎn)換函數(shù)的主體。也許我們將從一些最小的轉(zhuǎn)換和派生開始。我們將用一些“這些是正確的？”的問題來繼續(xù)探索。當(dāng)我們發(fā)現(xiàn)不工作時(shí)，我們會(huì)從中取出一些。

我們的整體處理過程如下所示：

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)) )

請(qǐng)注意語句主體的變化。convert（）函數(shù)產(chǎn)生我們確定的值。我們已經(jīng)在for循環(huán)中添加了一些額外的變量，我們不能100％確定。在更新convert（）函數(shù)之前，我們會(huì)看看它們是否有用（甚至是正確的）。

在減量方面，我們可以采取稍微不同的加工方式。我們需要重構(gòu)我們之前的例子，并把它變成一個(gè)生成器函數(shù)。

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

接著用一個(gè)yield代替了print（）。

這是重構(gòu)的另一部分。

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

理想情況下，我們所有的編程都是這樣的。我們使用生成器函數(shù)來生成數(shù)據(jù)。數(shù)據(jù)的最終顯示保持完全分離。這使我們可以更自由地重構(gòu)和改變處理。

現(xiàn)在我們可以做一些事情，例如將行收集到Counter（）對(duì)象中，或者可能計(jì)算一些統(tǒng)計(jì)信息。我們可以使用defaultdict（list）按服務(wù)對(duì)行進(jìn)行分組。

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()) )

我們決定在這里創(chuàng)建具體的列表對(duì)象。我們可以使用itertools按服務(wù)分組響應(yīng)時(shí)間。它看起來像是正確的函數(shù)式編程，但是這種實(shí)施在Pythonic函數(shù)式編程形式中指出了一些限制。要么我們必須對(duì)數(shù)據(jù)進(jìn)行排序（創(chuàng)建列表對(duì)象），要么在分組數(shù)據(jù)時(shí)創(chuàng)建列表。為了做好幾個(gè)不同的統(tǒng)計(jì)，通過創(chuàng)建具體的列表來分組數(shù)據(jù)通常更容易。

我們現(xiàn)在正在做兩件事情，而不是簡單地打印行對(duì)象。

創(chuàng)建一些局部變量，如svc和m。我們可以很容易地添加變化或其他措施。

使用沒有參數(shù)的vars（）函數(shù)，它會(huì)從局部變量中創(chuàng)建一個(gè)字典。

這個(gè)使用vars（）而沒有參數(shù)的行為就像locals（）一樣是一個(gè)方便的技巧。它允許我們簡單地創(chuàng)建我們想要的任何局部變量，并將它們包含在格式化輸出中。我們可以侵入我們認(rèn)為可能相關(guān)的各種統(tǒng)計(jì)方法中。

既然我們的基本處理循環(huán)是針對(duì)converted_log（“somefile.csv”）中的行，我們可以通過一個(gè)小小的，易于修改的腳本探索很多處理選擇。我們可以探索一些假設(shè)來確定為什么某些RESTful API處理速度慢，而其他處理速度則很快。

問答 
如何在Python中分析內(nèi)存使用情況？
相關(guān)閱讀
基于Python實(shí)現(xiàn)的微信好友數(shù)據(jù)分析
Python數(shù)據(jù)分析和數(shù)據(jù)挖掘?qū)W習(xí)路線圖
一文入門Python數(shù)據(jù)分析庫Pandas

此文已由作者授權(quán)騰訊云+社區(qū)發(fā)布，原文鏈接：https://cloud.tencent.com/dev...