使用Luigi來搭建Data Pipeline

爲何須要Luigi?

常見的資料處理流程可租略分紅如下幾步:

1. 預處理: 整合不一樣來源的數據, 篩選相關的數據, 清洗數據, 標準化(normalization) 等處理
2. 模型訓練: 透過一些機器學習算法來創建模型
3. 呈現或預測: 將訓練獲得的模型用於分析模式或是預測

剛開始你的資料處理流程多是像這樣依序地執行腳本

$ python get_data.py
$ python clean_data.py
$ python normalize.py
$ python train.py
$ python predict.py

稍微懶一點的人因而會把全部腳本打包進一個腳本run_all.py

if __name__ == '__main__':
    get_data()
    clean_data()
    normalize()
    train()
    predict()

$ python run_all.py

可是若是想要作好錯誤處理(error handling), 腳本可能就會變成像這樣嘮叨的代碼

try:
    get_data()
    try:
        clean_data()
        try:
            # ...
        except MoreErrors:
            # ...
    except CleanDataError as e:
        # handle CleanDataError
except GetDataError as e:
    # handle GetDataError

一旦步驟切分得更細或是步驟愈來愈多時, 錯誤處理變成一項繁瑣的工做, 這樣的重複的結構也使得代碼不易閱讀. 一些框架如Luigi和Airflow能將程序員從這項勞動中解放出來, 提高開發的效率,

什麼是Luigi?

Luigi是一套管理上述工做流程的python框架,讓開發人員專一在數據處理的邏輯,而沒必要重複的撰寫經常使用的代碼, 例如前面舉例的錯誤處理. 另外, 若是某個數據處理的步驟發生了錯誤或是更動,使用Luigi能夠避免從新計算不被影響的結果.

安裝Luigi

1. 穩定版本 $ pip install luigi
2. 最新版本 $ pip install git+https://github.com/spotify/luigi.git

運行Luigi

1. 執行 Luigi daemon $ luigid
2. 開啓瀏覽器前往 http://localhost:8082