摘要：在中運行不同的實驗似乎比試圖在中進行這種探索性的操作更有效。理論上，我們可以在中做很多的探索。我們如何繼續第一步是獲取格式的原始數據。這些列將包含來自使用該代理鍵的一個請求的一行數據。這是重構的另一部分。數據的最終顯示保持完全分離。

歡迎大家前往騰訊云+社區，獲取更多騰訊海量技術實踐干貨哦~

這里有一些技巧來處理日志文件提取。假設我們正在查看一些Enterprise Splunk提取。我們可以用Splunk來探索數據。或者我們可以得到一個簡單的提取并在Python中擺弄這些數據。

在Python中運行不同的實驗似乎比試圖在Splunk中進行這種探索性的操作更有效。主要是因為我們可以無所限制地對數據做任何事。我們可以在一個地方創建非常復雜的統計模型。

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

但是，使用Splunk需要假設我們知道我們正在尋找什么。在很多情況下，我們不知道我們在尋找什么：我們正在探索。可能會有一些跡象表明，一些RESTful API處理速度很慢，但還不止于此。我們如何繼續？

第一步是獲取CSV格式的原始數據。怎么辦？

我們將首先用一些附加函數來包裝一個CSV.DictReader對象。

面向對象的純粹主義者會反對這個策略。 “為什么不擴展DictReader？”他們問。我沒有一個很好的答案。我傾向于函數式編程和組件的正交性。對于一個純粹的面向對象的方法，我們不得不使用更復雜的混合來實現這一點。

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

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

我們可以 - 在一定程度上 - 以有用的格式報告原始數據。如果我們想粉飾一下輸出，我們可以改變格式字符串。那就可能是“{主機：30s} {回復時間：8s} {來源：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）。有些人喜歡這個發生器表達式。

我們可以用一個適當的for語句和一個內部的yield語句，但是從一個小的東西里寫大的語句似乎沒有什么好處。

我們有很多選擇，因為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（）函數之前，我們會看看它們是否有用（甚至是正確的）。

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

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中分析內存使用情況？
相關閱讀
基于Python實現的微信好友數據分析
Python數據分析和數據挖掘學習路線圖
一文入門Python數據分析庫Pandas

此文已由作者授權騰訊云+社區發布，原文鏈接：https://cloud.tencent.com/dev...