「Data Show」DS003 | 用進度條助您一臂之力

來自專欄 λ & τ20 人贊了文章

在具體的分析或者特徵工程之中，經常會遇到處理時間很久的問題，當然必要的優化是必須的。但是顯然，數據量上升，計算量過大後，處理時間是必須的此。時，如果有個可以幫助您查看任務進度的進度條，必定可以提高你抓住處理時間去做（磨）別（洋）事（工）。當然逐行列印是不錯的選擇，但在Jupyter notebook/JupyterLab中，這種實踐最大的問題是，列印過多，影響整個notebook的美觀程度。

在此探討的是5GB級別以下的數據（之上的Spark分析，有基於Zipplin的分散式任務精度條），主要環境是Jupyter下基於pandas包的分析和特徵工程任務。

一個極為簡單的例子

tqdm是基於Python的精度條模塊，裡面提供了簡單的代碼行進度條和基於ipywidgets的notebook內的進度條。由於現在tqdm相關模塊還在開發階段，可能會用到一些私有對象，之後正式版中可能具體API會有所變化。

當然，首先我們得載入模塊，在notebook中使用tqdm帶的基於Js顯示的進度條前，請務必檢查是否安裝ipywidgets模塊。

from tqdm import tqdm_notebook, _tqdm_notebook_tqdm_notebook.tqdm_notebook.pandas()

其中第一行載入的兩個方法的作用分別是：

• tqdm_notebook：用來包裝任何可以iterable的對象，在使用其元素進行運算結束後統計時間。
• _tqdm_notebook：其中含有模塊可以處理pandas的對象。

第二行則是重載pandas裡面的對象，提供可以展示精度條的方法。

下面我們可以嘗試直接使用tqdm_notebook包裹iterable對象來展示進度條，效果如下：

a = list(range(1, 10000))b = range(1, 10000)_ = [(lambda x: x+1)(i) for i in tqdm_notebook(a)]_ = [(lambda x: x+1)(i) for i in tqdm_notebook(b)]

當然如果僅僅是使用range也可以使用tqdm自帶的tnrange：

from tqdm._tqdm_notebook import tnrange_ = [(lambda x: x+1)(i) for i in tnrange(1, 10000)]

效果如下：

命名和深度

在一些場合，可能寫需要多個層級的迭代，此時，我們可以通過命名每個層級的迭代器來實現這個個效果。使用desc參數即可：

for i in tnrange(1, 10, desc=i Loop): for j in tnrange(1, 10000, desc=j Loop): i+j

當然，如果遇到Loop過多時，可能會依舊出現列印過多的困擾。此時leave參數是一個不錯的推薦。

for i in tqdm_notebook(range(100), desc=i-Loop): for j in tqdm_notebook(range(10000), desc=j-Loop, leave=False): i+j

多進程的擴展

當然，在具體計算中，多進程往往是經常會需要的一類擴展（由於Python只能基於一個運算核心進行計算的限制），這時候線程的運算也是經常需要考量的方式。

在使用過程中，第一個需要注意的問題是，tqdm每次是在從iterable對象中取值時，進行更新，而如果在map之前的list中做進度條的包裹，是在未使用map的函數之前統計。所以在進度條完成時，可能還會有一段時間後才真的執行結束。

from multiprocessing import Pooldef f(x): return x**32p = Pool(5)_ = [i for i in p.imap(f, tnrange(1000))]

而一個更好的處理是在使用後標記時間，使用multiprocessing.Pool.imap作為迭代對象，但這個問題是tqdm無法識別具體數量，此時，指定tqdm的迭代次數total即可。

_ = [i+1 for i in tqdm_notebook(p.imap(f, range(1000)))]

_ = [i for i in tqdm_notebook(p.imap(f, range(3, 1000)), total=997)]

pandas中使用

pandas中的使用，也是非常簡單，在重載命令執行後，Serires、DataFrame、GroupBy對象都會擁有progress_apply方法，用法和apply一致，直接可以調取進度條。

實戰：複雜場景中的使用

最後，我們結合一下之前的多線程和pandas操作，處理更大規模的數據。基本思路是，將DataFrame拆成若干組分，最後通過pandas.concat合併起結果。

def parallelize_dataframe(df, func, n_jobs=3, split_num=10): ## 拆分數據表 df_split = np.array_split(df, split_num) pool = Pool(n_jobs) df_list = [] ## map操作 for df_element in tqdm_notebook(pool.imap(func, df_split), total=10000): df_list.append(df_element) ## reduce操作 df = pd.concat(df_list) ## 關閉進程 pool.close() pool.join() return df

以上實現了基本的基於tqdm顯示處理進度的操作。使用方法如下：

def apply_add_1(df): return df.apply(lambda row: row[sepal_length]+1, axis=1)_ = parallelize_dataframe(iris_df, apply_add_1)

結語

查看了一下進度條，這次預處理我還要花一小時，可以先去沖杯咖啡了。