复古的金色钥匙在黑暗中飞翔。这幅画面象征着探索未知领域的决心和勇气。

有时，作为一名数据科学家，我们常常忘记了最初的追求。我们首先是开发者，然后是研究人员，最后才是数学家。我们的首要任务是迅速找到没有bug的解决方案。

编写模型并不意味着我们就是神。这绝不是编写劣质代码的理由。

自从我开始学习机器学习以来，我犯了许多错误。因此，我想把我认为在机器学习工程中最常用且最重要的技能分享给大家。在我看来，这也是当前行业中最缺乏的技能之一。

我称之为不懂软件的数据科学家，因为很大一部分人并没有系统地学习过计算机科学课程。我自己也有这样的经历。

如果必须选择一个伟大的数据科学家和一个伟大的机器学习工程师，我会选择后者。

下面是我的分享：

学习编写抽象类

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

```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 Predictor(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 参数来限制数据加载量。这种方法非常适合本地配置不足的情况。

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

预测失败：成熟的开发人员的标志

始终检查数据中的缺失值，因为这些数据可能会导致问题。即使你当前的数据没有缺失值，也不能保证未来不会有。因此，务必注意这个问题。

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

显示处理进度

在处理大数据时，了解当前进度非常重要。这里提供两种方法：tqdm 和 fastprogress。

使用 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 += char ```

使用 fastprogress

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

mb = masterbar(range(10)) for i in mb: for j in progressbar(range(100), parent=mb): time.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(func): """装饰器，用于记录函数执行时间""" @wraps(func) def wrapper(args, *kwargs): start = time.time() result = func(args, *kwargs) end = time.time() print(f'Function {func.name} took: {end - start:.2f} sec') return result return wrapper ```

不要在云上浪费资源

没有人喜欢浪费云资源的工程师。有些实验可能会持续数小时，因此需要跟踪并在完成后关闭云实例。使用 try 和 except 来包裹主函数，一旦发生异常，服务器将自动停止。

```python import os

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

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

创建和保存报告

在建模过程中，深入的见解往往来自于对误差和度量的分析。确保为本人和下属创建并保存格式正确的报告。

```python import json import os from sklearn.metrics import accuracyscore, classificationreport, confusionmatrix, f1score, fbeta_score

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

def savemetrics(metrics, 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') with open(path, 'w') as f: json.dump(metrics, f, indent=4) ```

写出一手好API

结果不好，一切都不好。你可以做好数据清理和建模，但如果最后的API设计不当，仍然可能导致混乱。以下是一些建议，适用于经典机器学习和深度学习部署，在较低负载下（例如每分钟1000次请求）。

推荐组合：FastAPI + Uvicorn + Gunicorn

• 快速：使用FastAPI编写API，因为它非常快。原因可以参考这篇文章。
• 文档：在FastAPI中编写API会自动生成免费的文档和测试端点，位于 http://url/docs。
• worker：使用Gunicorn服务器部署API，因为它支持启动多个worker，建议保留至少两个worker。

运行以下命令部署四个worker：

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

根据实际负载情况调整worker数量。

希望以上内容对你有所帮助！