数据科学家的初心与职责

作为数据科学家，我们往往容易忘记自己的初衷。我们首先是开发者，其次是研究人员，最后才是数学家。我们的主要任务是迅速找到没有bug的解决方案。虽然我们有能力构建模型，但这并不意味着我们可以随心所欲地编写代码。编写劣质代码是不可取的。

从我开始学习机器学习以来，我犯了很多错误。因此，我想分享一些在机器学习工程中常用的技能，这些技能也是目前行业中最缺乏的。我认为，那些不懂软件的数据科学家通常没有系统地学习过计算机科学课程。我自己也曾经历过这样的阶段。

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

接下来是我想要分享的内容。

学习编写抽象类

一旦开始编写抽象类，你就能体验到其带来的好处。抽象类强制子类使用相同的方法和方法名称。这样可以避免在同一个项目中因不同的人定义不同的方法而导致的混乱。以下是一个简单的例子：

```python 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 import random import numpy as np import torch

def setseed(seed): random.seed(seed) np.random.seed(seed) torch.manualseed(seed) if torch.cuda.isavailable(): torch.cuda.manualseed_all(seed) ```

先加载少量数据

如果你的数据量非常大，并且你正在处理数据清理或建模等后续步骤，可以使用 nrows 参数来避免每次加载大量数据。当你只想测试代码而不是实际运行整个程序时，这种方法非常有用。

```python import pandas as pd

加载前1000行数据

ftrain = pd.readcsv('train.csv', nrows=1000) ```

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

总是检查数据中的缺失值（NA），因为这些数据可能会导致问题。即使当前的数据没有缺失值，也不能保证将来不会出现。因此，始终要注意这个问题。

```python import pandas as pd

检查数据中的缺失值

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

显示处理进度

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

方法一：使用 tqdm

```python from tqdm import tqdm import time

使用 tqdm 进度条

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(f): """装饰器，用于记录函数执行时间""" def wrapper(args, *kwargs): start = time.time() result = f(args, *kwargs) end = time.time() print(f'function:{f.name} took: {end - start:.2f} sec') 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(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(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') with open(path, 'w') as f: json.dump(metrics, f, indent=4) ```

编写高质量的API

结果不好，一切都白费。即使你能做好数据清理和建模，但如果API设计不佳，最终也会出现问题。以下是一些关于经典机器学习和深度学习部署的最佳实践：

• 使用 FastAPI: 它是目前最快的框架之一，自带文档和测试端点。
• 使用 Gunicorn: 启动多个worker，至少保留两个worker。

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

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

以上内容涵盖了我在机器学习工程中积累的一些经验，希望对你有所帮助。