有时候，作为数据科学家，我们会忘记公司付钱让我们干什么。我们首先是开发人员，然后是研究人员，然后可能是数学家。我们的首要责任是快速开发无bug的解决方案。
我们能做模型并不意味着我们就是神。它没有给我们写垃圾代码的自由。
从一开始，我就犯了很多错误，我想和大家分享一下我所看到的ML工程中最常见的技能。在我看来，这也是目前这个行业最缺乏的技能。
我称他们为“软件文盲”，因为他们中的很多人都是非计算机科学课程学习平台(Coursera)的工程师。我自己曾经就是😅
如果要在一个伟大的数据科学家和一个伟大的ML工程师之间招聘，我会选择后者。让我们开始吧。
一旦你开始编写抽象类，你就会知道它能给你的代码库带来多大的清晰度。它们执行相同的方法和方法名称。如果很多人都在同一个项目上工作，每个人都会开始使用不同的方法。这可能会造成无效率的混乱。
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."""
实验的可重复性是非常重要的，而种子是我们的敌人。抓住它，否则会导致不同的训练/测试数据分割和不同的权值初始化神经网络。这导致了不一致的结果。
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)
如果你的数据太大，而你的工作是代码的后面的部分，如清理数据或建模，那么可以使用nrows来避免每次加载巨大的数据。当你只想测试代码而不实际运行整个代码时，请使用此方法。
当你的本地PC配置无法加载所有的数据的时候，但你又喜欢在本地开发时，这是非常适用的，
df_train = pd.read_csv(‘train.csv’, nrows=1000)
一定要检查数据中的NA，因为这些会给你以后带来问题。即使你当前的数据没有，这并不意味着它不会在未来的再训练循环中发生。所以无论如何😆继续检查。
print(len(df))
df.isna().sum()
df.dropna()
print(len(df))
当你在处理大数据时，知道它将花费多少时间以及我们在整个处理过程中的位置肯定会让你感觉很好。
选项 1 — tqdm
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
选项 2 — fastprogress
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}.')
如果你使用过pandas，你就会知道有时它有多慢 —— 尤其是groupby。不用打破头寻找“伟大的”解决方案加速，只需使用modin改变一行代码就可以了。
import modin.pandas as pd
不是所有的函数都是生而平等的
即使整个代码都能工作，也不意味着你写的代码很棒。一些软件bug实际上会使你的代码变慢，所以有必要找到它们。使用这个装饰器来记录函数的时间。
import time
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.2f sec' % (f.__name__, end - start))
return result
return wrapper
没有人喜欢浪费云资源的工程师。
我们的一些实验可以持续几个小时。很难跟踪它并在它完成时关闭云实例。我自己也犯过错误，也见过有人把实例开了好几天。
这种情况发生在星期五，离开后，周一才意识到😆
只要在执行结束时调用这个函数，你的屁股就再也不会着火了!!
但是将主代码包装在try中，此方法也包装在except中 —— 这样如果发生错误，服务器就不会继续运行。是的，我也处理过这些情况😅
让我们更负责任一点，不要产生二氧化碳。😅
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)
在建模的某个特定点之后，所有伟大的见解都只来自错误和度量分析。确保为自己和你的管理层创建和保存格式良好的报告。
😆管理层喜欢报告，对吗？😆
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)
所有的结果都是坏的。All that ends bad is bad.
你可以进行很好的数据清理和建模，但最终仍可能造成巨大的混乱。我与人打交道的经验告诉我，许多人不清楚如何编写好的api、文档和服务器设置。我很快会写另一篇关于这个的文章，但是让我开始吧。
下面是在不太高的负载下(比如1000/min)部署经典的ML和DL的好方法。
fasbut + uvicorn
- Fastest — 使用fastapi编写API，因为它很快。
- Documentation — 用fastapi写API让我们不用操心文档。
使用4个worker运行这些命令进行部署。通过负载测试优化workers的数量。
pip install fastapi uvicorn
uvicorn main:app --workers 4 --host 0.0.0.0 --port 8000
