創(chuàng)建數(shù)據(jù)處理管道

問(wèn)題

你想以數(shù)據(jù)管道 (類似 Unix 管道) 的方式迭代處理數(shù)據(jù)。比如，你有個(gè)大量的數(shù)據(jù)需要處理，但是不能將它們一次性放入內(nèi)存中。

解法

生成器函數(shù)是一個(gè)實(shí)現(xiàn)管道機(jī)制的好辦法。為了演示，假定你要處理一個(gè)非常大的日志文件目錄：

foo/access-log-012007.gzaccess-log-022007.gzaccess-log-032007.gz...access-log-012008 bar/access-log-092007.bz2...access-log-022008

假設(shè)每個(gè)日志文件包含這樣的數(shù)據(jù)：

124.115.6.12 - - [10/Jul/2012:00:18:50 -0500] "GET /robots.txt ..." 200 71 210.212.209.67 - - [10/Jul/2012:00:18:51 -0500] "GET /ply/ ..." 200 11875 210.212.209.67 - - [10/Jul/2012:00:18:51 -0500] "GET /favicon.ico ..." 404 369 61.135.216.105 - - [10/Jul/2012:00:20:04 -0500] "GET /blog/atom.xml ..." 304 - ...

為了處理這些文件，你可以定義一個(gè)由多個(gè)執(zhí)行特定任務(wù)獨(dú)立任務(wù)的簡(jiǎn)單生成器函數(shù)組成的容器。就像這樣：

import os import fnmatch import gzip import bz2 import re def gen_find(filepat, top):'''Find all filenames in a directory tree that match a shell wildcard pattern'''for path, dirlist, filelist in os.walk(top):for name in fnmatch.filter(filelist, filepat):yield os.path.join(path,name)def gen_opener(filenames):'''Open a sequence of filenames one at a time producing a file object.The file is closed immediately when proceeding to the next iteration.'''for filename in filenames:if filename.endswith('.gz'):f = gzip.open(filename, 'rt')elif filename.endswith('.bz2'):f = bz2.open(filename, 'rt')else:f = open(filename, 'rt')yield ff.close()def gen_concatenate(iterators):'''Chain a sequence of iterators together into a single sequence.'''for it in iterators:yield from itdef gen_grep(pattern, lines):'''Look for a regex pattern in a sequence of lines'''pat = re.compile(pattern)for line in lines:if pat.search(line):yield line

現(xiàn)在你可以很容易的將這些函數(shù)連起來(lái)創(chuàng)建一個(gè)處理管道。比如，為了查找包含單詞 python 的所有日志行，你可以這樣做：

lognames = gen_find('access-log*', 'www') files = gen_opener(lognames) lines = gen_concatenate(files) pylines = gen_grep('(?i)python', lines) for line in pylines:print(line)

如果將來(lái)的時(shí)候你想擴(kuò)展管道，你甚至可以在生成器表達(dá)式中包裝數(shù)據(jù)。比如，下面這個(gè)版本計(jì)算出傳輸?shù)淖止?jié)數(shù)并計(jì)算其總和。

lognames = gen_find('access-log*', 'www') files = gen_opener(lognames) lines = gen_concatenate(files) pylines = gen_grep('(?i)python', lines) bytecolumn = (line.rsplit(None,1)[1] for line in pylines) bytes = (int(x) for x in bytecolumn if x != '-') print('Total', sum(bytes))

討論

以管道方式處理數(shù)據(jù)可以用來(lái)解決各類其他問(wèn)題，包括解析，讀取實(shí)時(shí)數(shù)據(jù)，定時(shí)輪詢等。

為了理解上述代碼，重點(diǎn)是要明白 yield 語(yǔ)句作為數(shù)據(jù)的生產(chǎn)者而 for 循環(huán)語(yǔ)句作為數(shù)據(jù)的消費(fèi)者。當(dāng)這些生成器被連在一起后，每個(gè) yield 會(huì)將一個(gè)單獨(dú)的數(shù)據(jù)元素傳遞給迭代處理管道的下一階段。在例子最后部分， sum() 函數(shù)是最終的程序驅(qū)動(dòng)者，每次從生成器管道中提取出一個(gè)元素。

這種方式一個(gè)非常好的特點(diǎn)是每個(gè)生成器函數(shù)很小并且都是獨(dú)立的。這樣的話就很容易編寫和維護(hù)它們了。很多時(shí)候，這些函數(shù)如果比較通用的話可以在其他場(chǎng)景重復(fù)使用。并且最終將這些組件組合起來(lái)的代碼看上去非常簡(jiǎn)單，也很容易理解。

使用這種方式的內(nèi)存效率也不得不提。上述代碼即便是在一個(gè)超大型文件目錄中也能工作的很好。事實(shí)上，由于使用了迭代方式處理，代碼運(yùn)行過(guò)程中只需要很小很小的內(nèi)存。

在調(diào)用 gen concatenate() 函數(shù)的時(shí)候你可能會(huì)有些不太明白。這個(gè)函數(shù)的目的是將輸入序列拼接成一個(gè)很長(zhǎng)的行序列。 itertools.chain() 函數(shù)同樣有類似的功能，但是它需要將所有可迭代對(duì)象最為參數(shù)傳入。在上面這個(gè)例子中，你可能會(huì)寫類似這樣的語(yǔ)句 lines = itertools.chain(*files) ，使得 gen_opener() 生成器能被全部消費(fèi)掉。但由于 gen_opener() 生成器每次生成一個(gè)打開過(guò)的文件，等到下一個(gè)迭代步驟時(shí)文件就關(guān)閉了，因此 china() 在這里不能這樣使用。上面的方案可以避免這種情況。

gen_concatenate() 函數(shù)中出現(xiàn)過(guò) yield from 語(yǔ)句，它將 yield 操作代理到父生成器上去。語(yǔ)句 yield from it 簡(jiǎn)單的返回生成器 it 所產(chǎn)生的所有值。關(guān)于這個(gè)我們?cè)?4.14 小節(jié)會(huì)有更進(jìn)一步的描述。

最后還有一點(diǎn)需要注意的是，管道方式并不是萬(wàn)能的。有時(shí)候你想立即處理所有數(shù)據(jù)。然而，即便是這種情況，使用生成器管道也可以將這類問(wèn)題從邏輯上變?yōu)楣ぷ髁鞯奶幚矸绞健?/p>

David Beazley 在他的 Generator Tricks for Systems Programmers 教程中對(duì)于這種技術(shù)有非常深入的講解。可以參考這個(gè)教程獲取更多的信息。

總結(jié)

以上是生活随笔為你收集整理的《Python Cookbook 3rd》笔记（4.13）：创建数据处理管道的全部?jī)?nèi)容，希望文章能夠幫你解決所遇到的問(wèn)題。

如果覺得生活随笔網(wǎng)站內(nèi)容還不錯(cuò)，歡迎將生活随笔推薦給好友。