原文：https://medium.com/modern-nlp/10-great-ml-practices-for-python-developers-b089eefc18fc
導語
有時候作為一名數據科學家，我們可能會忘記自己的主要職責。我們首先是開發者，接著是研究者，最後還可能是數學家。因此，我們最重要的責任就是快速開發出一個沒有錯誤的解決方案。
只因為我們能創建模型，並不意味著我們就是神，這不會帶給我們可以寫出糟糕的代碼的自由。
從我開始進入這個領域，我曾經犯下很多錯誤，並考慮分享一些我所知道的對於機器學習工程開發中最常用的技能。在我看來，這些也是工業界目前最缺乏的一些技能。
如果必須在一個優秀的數據科學家和一個優秀的機器學習工程師中選擇招聘一個，我會選擇後者。
下面就開始介紹 10 個實用的機器學習建議吧。
1. 學會寫抽象類
當你開始寫抽象類的時候，你就知道它可以讓你的代碼庫變得很清晰明瞭，它們會強制採用一樣的方法和方法名字。如果同個項目有多個人參與，並且每個人都用不同方法，那會產生不必要的混亂情況。
下面是一個代碼例子：
<code>import os
from abc import ABCMeta, abstractmethod
class DataProcessor(metaclass=ABCMeta):
"""Base processor to be used for all preparation."""
def __init__(self, input_directory, output_directory):
self.input_directory = input_directory
self.output_directory = output_directory
@abstractmethod
def read(self):
"""Read raw data."""
@abstractmethod
def process(self):
"""Processes raw data. This step should create the raw dataframe with all the required features. Shouldn't implement statistical or text cleaning."""
@abstractmethod
def save(self):
"""Saves processed data."""
class Trainer(metaclass=ABCMeta):
"""Base trainer to be used for all models."""
def __init__(self, directory):
self.directory = directory
self.model_directory = os.path.join(directory, 'models')
@abstractmethod
def preprocess(self):
"""This takes the preprocessed data and returns clean data. This is more about statistical or text cleaning."""
@abstractmethod
def set_model(self):
"""Define model here."""
@abstractmethod
def fit_model(self):
"""This takes the vectorised data and returns a trained model."""
@abstractmethod
def generate_metrics(self):
"""Generates metric with trained model and test data."""
@abstractmethod
def save_model(self, model_name):
"""This method saves the model in our required format."""
class Predict(metaclass=ABCMeta):
"""Base predictor to be used for all models."""
def __init__(self, directory):
self.directory = directory
self.model_directory = os.path.join(directory, 'models')
@abstractmethod
def load_model(self):
"""Load model here."""
@abstractmethod
def preprocess(self):
"""This takes the raw data and returns clean data for prediction."""
@abstractmethod
def predict(self):
"""This is used for prediction."""
class BaseDB(metaclass=ABCMeta):
""" Base database class to be used for all DB connectors."""
@abstractmethod
def get_connection(self):
"""This creates a new DB connection."""
@abstractmethod
def close_connection(self):
"""This closes the DB connection."""/<code>
2. 固定好隨機種子
實驗的可復現是非常重要的一件事情，而隨機種子可能會造成實驗結果無法復現。因此必須固定好隨機種子，否則會導致不同的訓練集和測試集，以及神經網絡的不同初始化權重，這些都會導致不一樣的實驗結果。
<code>def set_seed(args):
random.seed(args.seed)
np.random.seed(args.seed)
torch.manual_seed(args.seed)
if args.n_gpu > 0:
torch.cuda.manual_seed_all(args.seed)/<code>
3. 先展示少量數據
如果你的數據量非常大，然後你接下來要做的工作是類似於清洗數據或者建模，那麼可以每次採用少量的數據來避免一次加載大量的數據。當然這個做法的前提是你只是希望測試代碼，而不是實際開始實現相應的工作。
這個做法非常實用，特別是你本地電腦的配置不足以加載全部數據集的時候，但你又想在本地電腦採用 Jupyter/ VS code/ Atom 做實驗。
代碼例子如下：
<code>df_train = pd.read_csv(‘train.csv’, nrows=1000)/<code>
4. 預估可能的錯誤（一個成熟開發者的標誌）
記得每次都要檢查數據是否存在空數據（NA)，因為這將帶來代碼出錯。當然，即便當前數據中不存在，這並不意味著在後續的訓練步驟中不會出現這種情況，所以需要保持這種檢查。
比如採用下述的代碼：
<code>print(len(df))
df.isna().sum()
df.dropna()
print(len(df))/<code>
5. 展示處理進度
當在處理大量的數據的時候，如果能夠知道總共需要的時間以及當前的處理進度是非常有幫助的。
這裡有幾種方法：
第一種方法：採用 tqdm 庫，代碼例子：
<code>from tqdm import tqdm
import time
tqdm.pandas()
df['col'] = df['col'].progress_apply(lambda x: x**2)
text = ""
for char in tqdm(["a", "b", "c", "d"]):
time.sleep(0.25)
text = text + char/<code>
第二種方法--fastprogress
<code>from fastprogress.fastprogress import master_bar, progress_bar
from time import sleep
mb = master_bar(range(10))
for i in mb:
for j in progress_bar(range(100), parent=mb):
sleep(0.01)
mb.child.comment = f'second bar stat'
mb.first_bar.comment = f'first bar stat'
mb.write(f'Finished loop {i}.')/<code>
效果如下圖所示：
6. Pandas 可能會很慢
如果你採用 pandas 庫，那麼你會發現有的時候它的速度會有多慢，特別是採用 groupby 函數的時候。不需要想盡辦法來尋找更好的加速方法，只需要修改一行代碼即可，如下所示，採用 modin 即可解決這個問題：
<code>import modin.pandas as pd/<code>
7. 計算函數運行的時間
不是所有的函數運行時間都是一樣的。
即便你的代碼都跑通了，但也不表示你寫出一手好代碼。有些軟性錯誤（soft-bugs）可能會導致你的代碼運行算的變慢，因此很有必要找到這些問題。可以採用下述裝飾器來打印函數的運行時間。
<code>import time
from functools import wraps
def timing(f):
"""Decorator for timing functions
Usage:
@timing
def function(a):
pass
"""
@wraps(f)
def wrapper(*args, **kwargs):
start = time.time()
result = f(*args, **kwargs)
end = time.time()
print('function:%r took: %2.4f sec' % (f.__name__, end - start))
return result
return wrapper/<code>
使用的例子如下所示：
<code>from time import sleep
@timing
def count_nums():
sleep(1)/<code>
結果如下所示：
8. 不要在雲服務器方面花太多錢
沒有人會喜歡一個浪費雲資源的工程師
有些實驗是需要跑上數十個小時的，這些實驗很難進行跟蹤並在實驗完成的時候關閉雲服務器的實例。我曾經犯過這樣的錯誤，同時也看到其他人也曾經因為這個忘記關閉好幾天。
這種情況通常發現在週五的時候，然後讓其一直運行到週一。
為了避免發生這種情況，可以在代碼的最後加入下列代碼。
不過，注意要主要代碼放在 try catch中進行捕獲異常，以防止發生錯誤。這種情況也是很可能發生的。
<code>import os
def run_command(cmd):
return os.system(cmd)
def shutdown(seconds=0, os='linux'):
"""Shutdown system after seconds given. Useful for shutting EC2 to save costs."""
if os == 'linux':
run_command('sudo shutdown -h -t sec %s' % seconds)
elif os == 'windows':
run_command('shutdown -s -t %s' % seconds)
/<code>
9. 創建並保存報告
在訓練模型後，所有的想法最終都是來自錯誤和評判指標的分析。因此需要創建並保存好一個格式不錯的報告，以便進行彙報。
下面是一個代碼例子：
<code>import json
import os
from sklearn.metrics import (accuracy_score, classification_report,
confusion_matrix, f1_score, fbeta_score)
def get_metrics(y, y_pred, beta=2, average_method='macro', y_encoder=None):
if y_encoder:
y = y_encoder.inverse_transform(y)
y_pred = y_encoder.inverse_transform(y_pred)
return {
'accuracy': round(accuracy_score(y, y_pred), 4),
'f1_score_macro': round(f1_score(y, y_pred, average=average_method), 4),
'fbeta_score_macro': round(fbeta_score(y, y_pred, beta, average=average_method), 4),
'report': classification_report(y, y_pred, output_dict=True),
'report_csv': classification_report(y, y_pred, output_dict=False).replace('\\n','\\r\\n')
}
def save_metrics(metrics: dict, model_directory, file_name):
path = os.path.join(model_directory, file_name + '_report.txt')
classification_report_to_csv(metrics['report_csv'], path)
metrics.pop('report_csv')
path = os.path.join(model_directory, file_name + '_metrics.json')
json.dump(metrics, open(path, 'w'), indent=4)/<code>
10. 寫好接口
你可以很好完成數據清理和訓練模型，但是也還可能在最後製造很大的錯誤，比如沒有寫好服務接口。我的經驗告訴我，很多人其實不知道如果寫出一個好的服務接口，文檔說明和服務安裝配置。後面我會寫另一篇文章介紹，但現在先簡單介紹一下。
下面是一個很好的用於經典的機器學習和深度學習的部署策略，但注意是請求數量不大的時候，比如每分鐘1000次。
一個組合：Fastapi + uvicorn + gunicorn
- Fastest--採用 fastapi 寫接口是非常快速的，正如[1]中的報告展示，以及原因可以參考[2]；
- 文檔-- fastapi有免費的官方文檔以及可以通過 http:url/docs 進行測試，並且這個鏈接可以自動生成以及隨著我們修改代碼而自動改變；
- Workers--採用 gunicorn 服務器部署接口是因為它具有開始多於 1 個 worker 的功能，並且你應該至少保持 2 個。
運行下列命令可以部署使用 4 個 wokers ，另外可以通過測試來優化這個 workers 的數量。
<code>pip install fastapi uvicorn gunicorn
gunicorn -w 4 -k uvicorn.workers.UvicornH11Worker main:app/<code>
一個運行例子如下圖所示：
參考
- https://www.techempower.com/benchmarks/#section=test&runid=7464e520-0dc2-473d-bd34-dbdfd7e85911&hw=ph&test=query&l=zijzen-7
- https://fastapi.tiangolo.com/benchmarks/
閱讀更多 算法猿的成長 的文章
