作为数据科学家，我们有时会忘记自己的初心。我们首先是开发者，然后是研究人员，最后才是数学家。我们的主要任务是快速找到没有错误的解决方案。

我们能够构建模型，但这并不意味着我们可以成为神。编写劣质代码不是理由。自从我开始学习机器学习以来，我犯了许多错误。因此，我想分享我认为在机器学习工程中最有用的技能之一，这也是当前行业中相对缺乏的技能。

很多人没有系统地学习计算机科学课程，导致他们被称为不懂软件的数据科学家。我自己也曾遇到过这种情况。

如果必须在伟大的数据科学家和伟大的机器学习工程师之间做出选择，我会选择后者。

接下来是我的分享。

学习编写抽象类

一旦开始编写抽象类，你就会发现它带来的好处。抽象类强制子类使用相同的方法和方法名称。当多人在同一项目上工作时，如果每个人都定义不同的方法，不仅没有必要，还会造成混乱。

```python import os from abc import ABCMeta, abstractmethod

class DataProcessor(metaclass=ABCMeta): """用于所有数据准备的基础处理器""" def init(self, inputdirectory, outputdirectory): self.inputdirectory = inputdirectory self.outputdirectory = outputdirectory

@abstractmethod
def read(self):
    """读取原始数据"""

@abstractmethod
def process(self):
    """处理原始数据。这一步应该创建包含所有必要特征的原始数据框。不应该实现统计或文本清理"""

@abstractmethod
def save(self):
    """保存处理后的数据"""

class Trainer(metaclass=ABCMeta): """用于所有模型的基础训练器""" def init(self, directory): self.directory = directory self.model_directory = os.path.join(directory, 'models')

@abstractmethod
def preprocess(self):
    """处理预处理数据并返回干净的数据。这更多涉及统计或文本清理"""

@abstractmethod
def set_model(self):
    """定义模型"""

@abstractmethod
def fit_model(self):
    """处理向量化数据并返回训练好的模型"""

@abstractmethod
def generate_metrics(self):
    """使用训练好的模型和测试数据生成指标"""

@abstractmethod
def save_model(self, model_name):
    """以所需格式保存模型"""

class Predict(metaclass=ABCMeta): """用于所有模型的基础预测器""" def init(self, directory): self.directory = directory self.model_directory = os.path.join(directory, 'models')

@abstractmethod
def load_model(self):
    """加载模型"""

@abstractmethod
def preprocess(self):
    """处理原始数据并返回干净的数据用于预测"""

@abstractmethod
def predict(self):
    """用于预测"""

class BaseDB(metaclass=ABCMeta): """用于所有数据库连接的基础数据库类""" @abstractmethod def get_connection(self): """创建新的数据库连接"""

@abstractmethod
def close_connection(self):
    """关闭数据库连接"""

```

固定随机数种子

实验的可重复性非常重要，随机数种子是一个需要特别注意的因素。如果设置不当，会导致训练/测试数据的不同分割以及神经网络中不同权重的初始化，从而影响结果的一致性。

python def set_seed(seed): random.seed(seed) np.random.seed(seed) torch.manual_seed(seed) if torch.cuda.is_available() and seed > 0: torch.cuda.manual_seed_all(seed)

先加载少量数据

如果你的数据集很大，并且你在处理数据清洗或建模等后续步骤时，可以使用 nrows 参数来避免每次都加载大量数据。这种方法非常适合在本地电脑配置不足，但你仍喜欢使用 Jupyter 或 VS Code 进行开发的场景。

python f_train = pd.read_csv('train.csv', nrows=1000)

预测失败（成熟的开发者的标志）

始终检查数据中的 NA（缺失值），因为这些数据可能会导致问题。即使当前数据没有缺失值，也不能保证未来不会出现。因此，无论何时都需要留意这个问题。

python print(len(df)) df.isna().sum() df.dropna(inplace=True) print(len(df))

显示处理进度

在处理大数据时，了解当前进度和还需要多少时间完成非常重要。以下是两种实现进度条的方法：

方法一：使用 tqdm

```python 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 ```

方法二：使用 fastprogress

```python from fastprogress.fastprogress import masterbar, progressbar from time import sleep

mb = masterbar(range(10)) for i in mb: for j in progressbar(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 慢的问题

如果你使用过 Pandas，你可能知道它有时会很慢，特别是在团队协作时。与其绞尽脑汁寻找加速方案，不如尝试使用 modin 改变一行代码。

python import modin.pandas as pd

记录函数的执行时间

并不是所有的函数都是平等的。即使所有代码都能正常运行，也不代表你写出了一手好代码。一些隐秘的错误实际上会使你的代码变慢。因此，有必要找到它们。使用以下装饰器来记录函数的执行时间。

```python import time

def timing(f): """用于记录函数执行时间的装饰器 使用示例： @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

```

不要在云上浪费资源

没有人喜欢浪费云资源的工程师。有些实验可能需要数小时才能完成，跟踪它们并在完成后关闭云实例是一项挑战。我也曾犯过这样的错误，甚至有人会连续几天不关机。

这种状况常常发生在我们周五下班时，留下一些东西运转，直到周一才意识到。只要在执行结束时调用这个函数，你的资源就不会浪费了！

```python import os

def run_command(cmd): return os.system(cmd)

def shutdown(seconds=0, os='linux'): """在给定秒数后关闭系统，适用于节省EC2成本 """ if os == 'linux': runcommand('sudo shutdown -h -t %s' % seconds) elif os == 'windows': runcommand('shutdown -s -t %s' % seconds) ```

创建和保存报告

在建模的某个特定点之后，所有深入的见解都来自对误差和度量的分析。确保为自己和下属创建并保存格式正确的报告。

```python import json import os

from sklearn.metrics import ( accuracyscore, classificationreport, confusionmatrix, f1score, fbeta_score )

def getmetrics(y, ypred, beta=2, averagemethod='macro', yencoder=None): if yencoder: y = yencoder.inversetransform(y) ypred = yencoder.inversetransform(ypred) return { 'accuracy': round(accuracyscore(y, ypred), 4), 'f1scoremacro': round(f1score(y, ypred, average=averagemethod), 4), 'fbetascoremacro': round(fbetascore(y, ypred, beta, average=averagemethod), 4), 'report': classificationreport(y, ypred, outputdict=True), 'reportcsv': classificationreport(y, ypred, outputdict=False).replace('n', 'rn') }

def savemetrics(metrics: dict, modeldirectory, filename): path = os.path.join(modeldirectory, filename + 'report.txt') with open(path, 'w') as f: f.write(metrics['reportcsv']) metrics.pop('reportcsv') path = os.path.join(modeldirectory, filename + '_metrics.json') json.dump(metrics, open(path, 'w'), indent=4) ```

写出一手好 API

结果不好，一切都将变得糟糕。你可以做好数据清洗和建模，但仍然会在最后制造混乱。通过我与人打交道的经验，我发现很多人不知道如何编写好的 API、文档和服务器设置。我将很快写一篇关于这方面的文章，但先让我简要分享一部分。

对于经典的机器学习和深度学习部署，适用于不太高的负载（例如每分钟1000次请求）。

来看看这个组合：FastAPI + Uvicorn + Gunicorn

• 最快的：使用 FastAPI 编写 API，因为它是最快的，原因详见这篇文章。
• 文档：在 FastAPI 中编写 API 提供了免费的 HTTP 文档和测试端点，当我们更改代码时，FastAPI 会自动更新这些文档。
• worker：使用 Gunicorn 服务器部署 API，因为它可以启动多个 worker，建议保留至少两个 worker。

运行以下命令以使用4个 worker 部署：

bash pip install fastapi uvicorn gunicorn gunicorn -w 4 -k uvicorn.workers.UvicornH11Worker main:app

可以根据负载测试优化 worker 数量。

希望这些技巧对你有所帮助！