MLOps. Советы и хитрости. 75 примеров кода

Вступление

MLOps, или операции машинного обучения, относятся к набору методов, которые оптимизируют разработку, развёртывание и обслуживание моделей машинного обучения. Эта статья призвана предоставить ценные советы и рекомендации по MLOps и обработке данных, охватывая широкий круг тем, таких как обучение моделей, предварительная обработка данных, оптимизация производительности, мониторинг и развёртывание моделей.

1. Обучение модели Dask-ML-Parallelize:

• Используйте Dask-ML для параллельного обучения и оценки моделей машинного обучения, эксплуатируя всю мощь вашего оборудования.
• С помощью Dask-ML вы можете быстро масштабировать свои рабочие нагрузки машинного обучения на несколько ядер, процессоров или даже кластеров, упрощая обучение и оценку больших моделей на больших наборах данных.

import dask_ml.model_selection as dcv
from sklearn.datasets import make_classification
from sklearn.svm import SVC

# Create a large dataset
X, y = make_classification(n_samples=100000, n_features=20, random_state=42)

# Define your model
model = SVC()

# Train your model in parallel using Dask-ML
params = {"C": dcv.Categorical([0.1, 1, 10]), "kernel": dcv.Categorical(["linear", "rbf"])}
search = dcv.RandomizedSearchCV(model, params, n_iter=10, cv=3)
search.fit(X, y)

Больше информации здесь.

2. Инструменты функций: Featuretools — это библиотека Python с открытым исходным кодом для автоматизированного проектирования функций, позволяющая создавать новые функции из необработанных данных с минимальными ручными усилиями.

import featuretools as ft

# Load your raw data into an entityset
es = ft.EntitySet(id="my_data")
es = es.entity_from_dataframe(entity_id="customers", dataframe=data, index="customer_id")

# Define relationships between entities
# ...

# Automatically generate new features
feature_matrix, feature_defs = ft.dfs(entityset=es, target_entity="customers", max_depth=2)

Больше информации здесь.

3. Tensorboard: TensorBoard — это мощный инструмент визуализации для TensorFlow, который позволяет вам отслеживать производительность вашей модели и различные показатели во время обучения и оценки.

import tensorflow as tf
from tensorflow.keras.callbacks import TensorBoard

# Define your model
model = tf.keras.Sequential([...])

# Compile your model
model.compile(optimizer='adam', loss='sparse_categorical_crossentropy', metrics=['accuracy'])

# Create a TensorBoard callback
tensorboard_callback = TensorBoard(log_dir="./logs")

# Train your model with the TensorBoard callback
model.fit(X_train, y_train, epochs=10, validation_data=(X_test, y_test), callbacks=[tensorboard_callback])

Больше информации здесь.

4. Tensorflow Serving:

• TensorFlow Serving — это высокопроизводительная система обслуживания моделей машинного обучения, разработанная для производственных сред.
• TensorFlow Serving поддерживает несколько моделей, управление версиями моделей и автоматическую загрузку и выгрузку моделей, что упрощает управление моделями машинного обучения и их обслуживание в масштабе.

# Save your TensorFlow model in the SavedModel format
model.save("my_model/1/")

# Install TensorFlow Serving
echo "deb [arch=amd64] http://storage.googleapis.com/tensorflow-serving-apt stable tensorflow-model-server tensorflow-model-server-universal" | sudo tee /etc/apt/sources.list.d/tensorflow-serving.list && \
curl https://storage.googleapis.com/tensorflow-serving-apt/tensorflow-serving.release.pub.gpg | sudo apt-key add -
sudo apt-get update && sudo apt-get install tensorflow-model-server

# Start TensorFlow Serving with your model
tensorflow_model_server --rest_api_port=8501 --model_name=my_model --model_base_path=$(pwd)/my_model

Больше информации здесь.

5. Автоматизируйте настройку гиперпараметров с помощью Optuna: Optuna — это мощная и гибкая библиотека оптимизации, которая может автоматически исследовать и оптимизировать гиперпараметры для ваших моделей машинного обучения.

import optuna
from sklearn.model_selection import cross_val_score
from sklearn.ensemble import RandomForestClassifier

def objective(trial):
    n_estimators = trial.suggest_int("n_estimators", 10, 200)
    max_depth = trial.suggest_int("max_depth", 3, 20)

    clf = RandomForestClassifier(n_estimators=n_estimators, max_depth=max_depth)
    score = cross_val_score(clf, X_train, y_train, cv=5).mean()

    return score

study = optuna.create_study(direction="maximize")
study.optimize(objective, n_trials=50)

best_params = study.best_params

Больше информации здесь.

6. SHAP: используйте SHAP (SHapley Additive exPlanations), чтобы объяснить выходные данные ваших моделей машинного обучения и получить представление об их поведении.

import shap
from sklearn.model_selection import train_test_split
from sklearn.ensemble import RandomForestRegressor

# Load and prepare your data
# ...

# Train a RandomForestRegressor
X_train, X_test, y_train, y_test = train_test_split(X, y, test_size=0.2)
model = RandomForestRegressor()
model.fit(X_train, y_train)

# Explain the model's predictions using SHAP
explainer = shap.Explainer(model)
shap_values = explainer(X_test)

# Plot the SHAP values for a single prediction
shap.plots.waterfall(shap_values[0])

Больше информации здесь.

7. Ray: Ray Tune — это мощная и гибкая библиотека для распределённой настройки гиперпараметров, позволяющая использовать всю мощь вашего оборудования для оптимизации моделей машинного обучения.

from ray import tune
from ray.tune.schedulers import ASHAScheduler
from sklearn.datasets import make_classification
from sklearn.model_selection import cross_val_score
from sklearn.ensemble import RandomForestClassifier

def train_model(config):
    n_estimators = config["n_estimators"]
    max_depth = config["max_depth"]
    
    clf = RandomForestClassifier(n_estimators=n_estimators, max_depth=max_depth)
    score = cross_val_score(clf, X_train, y_train, cv=3).mean()
    
    tune.report(mean_accuracy=score)

# Load your data
X, y = make_classification(n_samples=1000, n_features=20, random_state=42)

# Define the search space for hyperparameters
config = {
    "n_estimators": tune.randint(10, 200),
    "max_depth": tu:ne.randint(3, 20)
}

# Set up Ray Tune
scheduler = ASHAScheduler(metric="mean_accuracy", mode="max")
analysis = tune.run(train_model, config=config, scheduler=scheduler, num_samples=50)

# Get the best hyperparameters
best_params = analysis.best_config

Больше информации здесь.

8. Отслеживание экспериментов с помощью MLflow: MLflow позволяет сравнивать различные прогоны, воспроизводить предыдущие результаты и делиться своей работой с другими, что делает совместную работу и итерации более эффективными.

import mlflow
import mlflow.sklearn
from sklearn.datasets import load_iris
from sklearn.model_selection import train_test_split
from sklearn.ensemble import RandomForestClassifier
from sklearn.metrics import accuracy_score

# Load your data
iris = load_iris()
X, y = iris.data, iris.target
X_train, X_test, y_train, y_test = train_test_split(X, y, test_size=0.2)

# Train your model and log metrics with MLflow
with mlflow.start_run():
    clf = RandomForestClassifier(n_estimators=100, max_depth=10)
    clf.fit(X_train, y_train)
    
    train_accuracy = clf.score(X_train, y_train)
    test_accuracy = clf.score(X_test, y_test)

    mlflow.log_param("n_estimators", 100)
    mlflow.log_param("max_depth", 10)
    mlflow.log_metric("train_accuracy", train_accuracy)
    mlflow.log_metric("test_accuracy", test_accuracy)

    mlflow.sklearn.log_model(clf, "model")

Больше информации здесь.

9. Scikit-learn: Pipeline: Используйте Scikit-learn Pipeline для объединения нескольких шагов предварительной обработки и окончательной оценки.

from sklearn.pipeline import Pipeline
from sklearn.preprocessing import StandardScaler
from sklearn.linear_model import LogisticRegression

pipe = Pipeline([
    ("scaler", StandardScaler()),
    ("classifier", LogisticRegression())
])

pipe.fit(X_train, y_train)

10. Scikit-learn: Grid search: используйте GridSearchCV для настройки гиперпараметров.

from sklearn.model_selection import GridSearchCV

param_grid = {
    "classifier__C": [0.1, 1, 10],
    "classifier__penalty": ["l1", "l2"]
}

grid_search = GridSearchCV(pipe, param_grid, cv=5)
grid_search.fit(X_train, y_train)

11. Joblib: joblib — популярная библиотека для сохранения и загрузки моделей Scikit-learn. Используйте dump() для сохранения модели в файл и load() для восстановления модели из файла.

import joblib

# Save the model
joblib.dump(grid_search.best_estimator_, "model.pkl")

# Load the model
loaded_model = joblib.load("model.pkl")

12. Tensorflow: простая нейронная сеть. Используйте Keras API для определения простой нейронной сети с прямой связью с плотными (полностью связанными) слоями.

import tensorflow as tf

model = tf.keras.Sequential([
    tf.keras.layers.Dense(64, activation="relu", input_shape=(10,)),
    tf.keras.layers.Dense(32, activation="relu"),
    tf.keras.layers.Dense(1, activation="sigmoid")
])

model.compile(optimizer="adam", loss="binary_crossentropy", metrics=["accuracy"])

13. Ранняя остановка: фрагмент кода для ранней остановки

early_stopping = tf.keras.callbacks.EarlyStopping(monitor="val_loss", patience=3)

history = model.fit(X_train, y_train, epochs=100, validation_split=0.2, callbacks=[early_stopping])

14. Сохранение и загрузка модели Tensorflow: используйте метод save(), чтобы сохранить архитектуру модели, веса и состояние оптимизатора в один файл. Используйте load_model() для восстановления сохранённой модели из файла.

# Save the model
model.save("model.h5")

# Load the model
loaded_model = tf.keras.models.load_model("model.h5")

15. Dask: Распараллеливание операций. Используйте Dask для распараллеливания операций над большими наборами данных.

import dask.array as da

x = da.ones((10000, 10000), chunks=(1000, 1000))
y = x + x.T
z = y.sum(axis=0)
result = z.compute()

16. TPOT: Автоматизированное машинное обучение: TPOT (Tree-based Pipeline Optimization Tool) — это библиотека автоматизированного машинного обучения на основе генетических алгоритмов. Используйте TPOTClassifier или TPOTRegressor, чтобы оптимизировать конвейер машинного обучения для ваших данных.

from tpot import TPOTClassifier

tpot = TPOTClassifier(generations=5, population_size=20, verbosity=2)
tpot.fit(X_train, y_train)

Больше информации здесь.

17. Category Encoders: Category Encoders — это библиотека, предоставляющая различные методы кодирования для категориальных переменных, такие как целевое кодирование, горячее кодирование и порядковое кодирование.

import category_encoders as ce

encoder = ce.TargetEncoder()
X_train_encoded = encoder.fit_transform(X_train, y_train)
X_test_encoded = encoder.transform(X_test)

18. Imbalanced-learn: это библиотека, предоставляющая различные методы обработки несбалансированных наборов данных, такие как передискретизация, недостаточная выборка и комбинированные методы. Используйте соответствующий метод повторной выборки, такой как SMOTE, чтобы сбалансировать набор данных перед обучением модели.

from imblearn.over_sampling import SMOTE

smote = SMOTE()
X_resampled, y_resampled = smote.fit_resample(X_train, y_train)

Больше информации здесь.

19. Auto-sklearn: это библиотека автоматизированного машинного обучения, которая является оболочкой Scikit-learn, обеспечивая автоматический выбор модели и предварительной обработки. Используйте AutoSklearnClassifier или AutoSklearnRegressor для оптимизации данных конвейера машинного обучения.

from autosklearn.classification import AutoSklearnClassifier

auto_classifier = AutoSklearnClassifier(time_left_for_this_task=600)
auto_classifier.fit(X_train, y_train)

20. Scikit-learn: Column Transformer: Column Transformer позволяет применять различные этапы предварительной обработки к разным столбцам входных данных, что особенно полезно при работе со смешанными типами данных.

from sklearn.compose import ColumnTransformer
from sklearn.preprocessing import StandardScaler, OneHotEncoder

preprocessor = ColumnTransformer(
    transformers=[
        ("num", StandardScaler(), ["numerical_feature_1", "numerical_feature_2"]),
        ("cat", OneHotEncoder(), ["categorical_feature"]),
    ]
)

X_train_transformed = preprocessor.fit_transform(X_train)
X_test_transformed = preprocessor.transform(X_test)

21. RandomizedSearchCV — это альтернатива GridSearchCV, которая более эффективно выполняет поиск в пространстве параметров путём случайной выборки фиксированного числа настроек параметров. Определите распределение параметров как словарь, где ключи — это имена параметров (включая имя шага при использовании конвейера), а значения — это распределения, из которых выбираются значения параметров. Передайте модель (или конвейер) и распределение параметров в RandomizedSearchCV и подгоните данные.

from sklearn.model_selection import RandomizedSearchCV
from scipy.stats import uniform

param_dist = {
    "classifier__C": uniform(loc=0, scale=4),
    "preprocessor__num__with_mean": [True, False],
}

random_search = RandomizedSearchCV(pipe, param_dist, n_iter=10, cv=5, scoring="accuracy")
random_search.fit(X_train, y_train)

22. TensorFlow Data Validation: используйте TensorFlow Data Validation (TFDV) для проверки и изучения ваших данных.

import tensorflow_data_validation as tfdv

stats = tfdv.generate_statistics_from_csv(data_location="train.csv")
schema = tfdv.infer_schema(statistics=stats)
tfdv.display_schema(schema=schema)

23. TensorFlow Model Analysis. Используйте TensorFlow Model Analysis (TFMA) для оценки ваших моделей TensorFlow.

import tensorflow_model_analysis as tfma

eval_shared_model = tfma.default_eval_shared_model(
    eval_saved_model_path="path/to/saved_model"
)
results = tfma.run_model_analysis(
    eval_shared_model=eval_shared_model,
    data_location="test.tfrecords",
    file_format="tfrecords",
    slice_spec=[tfma.slicer.SingleSliceSpec()]
)
tfma.view.render_slicing_metrics(results)

24. TensorFlow Transform: используйте TensorFlow Transform (TFT) для предварительной обработки ваших данных для моделей TensorFlow.

import tensorflow_transform as tft

def preprocessing_fn(inputs):
    outputs = {}
    outputs["scaled_feature"] = tft.scale_to_z_score(inputs["numerical_feature"])
    outputs["one_hot_feature"] = tft.compute_and_apply_vocabulary(inputs["categorical_feature"])
    return outputs

25. TensorFlow Extended (TFX): используйте TensorFlow Extended (TFX) для создания сквозных конвейеров машинного обучения.

from tfx.components import CsvExampleGen, Trainer
from tfx.orchestration.experimental.interactive.interactive_context import InteractiveContext

context = InteractiveContext()

example_gen = CsvExampleGen(input_base="path/to/data")
context.run(example_gen)

trainer = Trainer(
    module_file="path/to/trainer_module.py",
    examples=example_gen.outputs["examples"],
    train_args=trainer_pb2.TrainArgs(num_steps=10000),
    eval_args=trainer_pb2.EvalArgs(num_steps=5000)
)
context.run(trainer)

26. CuPy: CuPy — это библиотека, которая предоставляет NumPy-подобный интерфейс для вычислений с ускорением на GPU. Используйте массивы CuPy, интерфейс которых аналогичен массивам NumPy, для выполнения вычислений на графическом процессоре. В CuPy доступны многие распространённые функции NumPy, позволяющие выполнять вычисления с ускорением на графическом процессоре со знакомым синтаксисом.

import cupy as cp

x = cp.array([1, 2, 3, 4, 5])
y = cp.array([6, 7, 8, 9, 10])

z = cp.dot(x, y)

27. RAPIDS — это набор библиотек с ускорением на GPU для обработки данных, включая cuDF (библиотека DataFrame с ускорением на GPU, аналогичная Pandas) и cuML (библиотека машинного обучения с ускорением на GPU, аналогичная Scikit-learn). Используйте cuDF DataFrames для выполнения задач по обработке данных на графическом процессоре и модели cuML для обучения и оценки моделей машинного обучения на графическом процессоре.

import cudf
import cuml

df = cudf.read_csv("data.csv")
kmeans_model = cuml.KMeans(n_clusters=5)
kmeans_model.fit(df)

28. FastAPI — это современная высокопроизводительная веб-инфраструктура для создания API-интерфейсов с помощью Python, особенно подходящая для моделей машинного обучения.

• Создайте экземпляр FastAPI и определите конечные точки API с помощью декораторов, таких как @app.post().
• Используйте uvicorn для запуска вашего приложения FastAPI, указав хост и порт.

from fastapi import FastAPI
import uvicorn

app = FastAPI()

@app.post("/predict")
async def predict(text: str):
    prediction = model.predict([text])
    return {"prediction": prediction}

if __name__ == "__main__":
    uvicorn.run(app, host="0.0.0.0", port=8000)

29. Streamlit — библиотека для быстрого создания интерактивных веб-приложений для машинного обучения и обработки данных с использованием только Python.

• Используйте простой API Streamlit для создания элементов пользовательского интерфейса, таких как ввод текста и ползунки, а также отображения вывода или визуализации.
• Запустите приложение Streamlit с помощью команды streamlit run app.py в терминале.

import streamlit as st

st.title("My Streamlit App")

input_text = st.text_input("Enter some text:")
st.write(f"You entered: {input_text}")

slider_value = st.slider("Select a value:", 0, 100, 50)
st.write(f"Slider value: {slider_value}")

Больше информации здесь.

30. Docker File: Создайте файл Docker, чтобы определить собственный образ Docker для вашего приложения машинного обучения.

FROM python:3.8

WORKDIR /app
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

COPY . .

CMD ["python", "app.py"]

• Используйте ключевое слово FROM, чтобы указать базовый образ, например официальный образ Python.
• Используйте ключевое слово WORKDIR, чтобы установить рабочий каталог для последующих инструкций.
• Используйте ключевое слово COPY для копирования файлов и каталогов из хост-системы в образ.
• Используйте ключевое слово RUN для выполнения команд в процессе сборки, таких как установка зависимостей.
• Используйте ключевое слово CMD, чтобы определить команду по умолчанию, которая будет запускаться при запуске контейнера.

31. Создайте образ докера:

docker build -t my_ml_app:latest .

32. Запустите контейнер Docker. Используйте команду docker run для создания и запуска контейнера Docker из образа. Используйте флаг -p, чтобы сопоставить порт хоста с портом контейнера, разрешив внешний доступ к службам, работающим внутри контейнера.

docker run -p 5000:5000 my_ml_app:latest

32. Kubernetes YAML Config File:

apiVersion: apps/v1
kind: Deployment
metadata:
  name: my-ml-app
spec:
  replicas: 3
  selector:
    matchLabels:
      app: my-ml-app
  template:
    metadata:
      labels:
        app: my-ml-app
    spec:
      containers:
      - name: my-ml-app-container
        image: my_ml_app:latest
        ports:
        - containerPort: 5000

---

apiVersion: v1
kind: Service
metadata:
  name: my-ml-app-service
spec:
  selector:
    app: my-ml-app
  ports:
    - protocol: TCP
      port: 80
      targetPort: 5000
  type: LoadBalancer

• Используйте apiVersion, kind и metadata, чтобы определить тип ресурса Kubernetes и метаданные.
• Используйте spec , чтобы определить желаемое состояние ресурса, например количество реплик, образов контейнеров и открытых портов.
• Используйте разделитель --- для определения нескольких ресурсов в одном файле, таких как развертывание и служба.

33. kubectl: используйте инструмент командной строки kubectl для управления кластером и ресурсами Kubernetes.

# Apply the Kubernetes configuration file
kubectl apply -f my_ml_app.yaml

# List all deployments
kubectl get deployments

# List all services
kubectl get services

# Scale the deployment
kubectl scale deployment my-ml-app --replicas=5

# Delete the deployment and service
kubectl delete -f my_ml_app.yaml

34. Организуйте свой проект:

my_ml_project/
|-- data/
|   |-- raw/
|   |-- processed/
|-- models/
|-- notebooks/
|-- src/
|   |-- features/
|   |-- models/
|   |-- utils/
|-- Dockerfile
|-- requirements.txt

• Используйте отдельные каталоги для данных, моделей, записных книжек и исходного кода.
• Кроме того, разделите каталоги, чтобы разделить необработанные и обработанные данные или различные типы модулей исходного кода.

35. Управление версиями моделей. Используйте инструменты управления версиями моделей, такие как DVC или MLflow, для отслеживания различных версий обученных моделей машинного обучения.

• Храните артефакты модели (например, веса, метаданные) в централизованной системе хранения, такой как Amazon S3 или Google Cloud Storage.
• Используйте инструмент управления версиями, чтобы отслеживать версии моделей, связанные с ними обучающие данные и гиперпараметры.
• Обеспечьте простое сравнение и воспроизводимость моделей, отслеживая показатели производительности и конфигурации обучения.

36. Автоматизированное тестирование:

• Используйте библиотеки тестирования, такие как unittest или pytest, для написания и запуска тестов.
• Тестируйте отдельные функции и классы с помощью модульных тестов и проверяйте взаимодействие между компонентами с помощью интеграционных тестов.
• Выполняйте комплексные тесты, чтобы убедиться, что вся система работает должным образом, включая обслуживание моделей и конечные точки API.

37. Papermill:

• Papermill позволяет параметризовать Jupyter Notebooks, вводя новые значения для определённых ячеек.
• Выполняйте Notebooks программно и создавайте отчёты с различными значениями параметров без ручного вмешательства.

import papermill as pm

pm.execute_notebook(
    input_path='input_notebook.ipynb',
    output_path='output_notebook.ipynb',
    parameters={'param1': 'value1', 'param2': 'value2'}
)

38. Управление средой: такие инструменты, как Conda или virtualenv, создают изолированные среды для проектов.

# Create a new Conda environment
conda create -n my_ml_env python=3.8

# Activate the environment
conda activate my_ml_env

# Install packages
conda install pandas scikit-learn

# Deactivate the environment
conda deactivate

39. Прогрессивная загрузка моделей. Загружайте большие модели порциями, чтобы уменьшить потребление памяти и повысить производительность.

import numpy as np
import pandas as pd
from sklearn.linear_model import LinearRegression

chunksize = 10000
model = LinearRegression()

for i, chunk in enumerate(pd.read_csv("large_dataset.csv", chunksize=chunksize)):
    X_chunk = chunk.drop("target", axis=1)
    y_chunk = chunk["target"]
    model.partial_fit(X_chunk, y_chunk)
    print(f"Processed chunk {i + 1}")

40. Особенности кодирования:

• Методы кодирования признаков преобразуют категориальные переменные в числовые представления, которые могут использовать модели машинного обучения.
• Горячее кодирование создаёт двоичные столбцы для каждой категории, в то время как целевое кодирование заменяет каждую категорию средним значением целевой переменной для этой категории.

import pandas as pd
from sklearn.preprocessing import OneHotEncoder

data = pd.DataFrame({"Category": ["A", "B", "A", "C"]})

encoder = OneHotEncoder()
encoded_data = encoder.fit_transform(data)

print(encoded_data.toarray())

41. Проверка данных. Проверяйте качество и согласованность ваших данных с помощью платформ проверки данных, таких как Great Expectations, Pandera или пользовательских функций проверки.

import pandera as pa
from pandera import DataFrameSchema, Column, Check

schema = DataFrameSchema({
    "age": Column(pa.Int, Check(lambda x: 18 <= x <= 100)),
    "income": Column(pa.Float, Check(lambda x: x >= 0)),
    "gender": Column(pa.String, Check(lambda x: x in ["M", "F", "Other"])),
})

# Validate your DataFrame
validated_df = schema.validate(df)

42. Управление версиями данных. Используйте инструменты управления версиями данных, такие как DVC или Pachyderm, чтобы отслеживать изменения в ваших наборах данных и обеспечивать воспроизводимость в различных экспериментах и версиях моделей.

# Initialize DVC in your project
dvc init

# Add your dataset to DVC
dvc add data/my_dataset

# Commit the changes to your Git repository
git add data/my_dataset.dvc .dvc/config
git commit -m "Add my_dataset to DVC"

43. Используйте хранилища функций. Внедрите хранилища функций, такие как Feast или Hopsworks, для хранения, управления и обслуживания функций моделей машинного обучения.

from feast import FeatureStore

# Initialize the feature store
store = FeatureStore(repo_path="path/to/your/feature_store")

# Fetch features for training
training_df = store.get_historical_features(
    entity_df=entity_df,
    feature_refs=["your_feature_name"]
).to_df()

# Fetch features for serving
feature_vector = store.get_online_features(
    feature_refs=["your_feature_name"],
    entity_rows=[{"your_entity_key": "your_value"}]
).to_dict()

Хранилища функций могут помочь вам централизовать управление вашими функциями, обеспечивая согласованность и уменьшая дублирование в разных моделях и экспериментах.

44. Масштабирование функций. Применяйте методы масштабирования функций, такие как масштабирование MinMax, стандартное масштабирование или нормализация, чтобы убедиться, что ваши функции имеют одинаковые масштабы и распределения.

from sklearn.datasets import load_iris
from sklearn.preprocessing import StandardScaler

X, y = load_iris(return_X_y=True)

# Scale features using standard scaling
scaler = StandardScaler()
X_scaled = scaler.fit_transform(X)

45. Уменьшение размерности. Применяйте методы уменьшения размерности, такие как PCA, t-SNE или UMAP, чтобы уменьшить количество функций в наборе данных, сохраняя при этом важные закономерности и взаимосвязи.

from sklearn.datasets import load_iris
from sklearn.decomposition import PCA

X, y = load_iris(return_X_y=True)

# Apply PCA to reduce the dimensionality of the dataset
pca = PCA(n_components=2)
X_reduced = pca.fit_transform(X)

46. Объединение в цепочки Pandas. Объединение операций Pandas в цепочки для создания более удобочитаемого и лаконичного кода для обработки данных.

import pandas as pd

data = pd.read_csv("my_data.csv")

# Chain Pandas operations
result = (
    data.query("age >= 30")
    .groupby("city")
    .agg({"salary": "mean"})
    .sort_values("salary", ascending=False)
)

47. Используйте функцию «pipe». Используйте функцию pipe для интеграции пользовательских функций или операций в рабочий процесс цепочки Pandas.

import pandas as pd

def custom_operation(df, column, value):
    return df[df[column] > value]

data = pd.read_csv("data.csv")

# Integrate custom operations using 'pipe'
result = (
    data.pipe(custom_operation, "age", 18)
    .groupby("city")
    .agg({"salary": "mean"})
    .sort_values("salary", ascending=False)
)

48. Встроенные функции построения графиков Pandas: используйте встроенные функции построения графиков Pandas для быстрой и простой визуализации данных.

import pandas as pd

data = pd.read_csv("my_data.csv")

# Create a bar plot of average salary by city
data.groupby("city")["salary"].mean().plot(kind="bar")

49. Визуализируйте отсутствующие данные с помощью Missingno: используйте библиотеку Missingno для визуализации отсутствующих данных в вашем наборе данных.

import pandas as pd
import missingno as msno

data = pd.read_csv("data.csv")

# Visualize missing data
msno.matrix(data)

50. Используйте базы данных SQL. Вы можете использовать библиотеку sqlite3 в Python для взаимодействия с базой данных SQLite. Например, вы можете создать таблицу в базе данных SQLite и вставить в неё некоторые данные:

import sqlite3

# Connect to an SQLite database
conn = sqlite3.connect('example.db')

# Create a table
conn.execute('CREATE TABLE IF NOT EXISTS my_table (id INTEGER PRIMARY KEY, name TEXT)')

# Insert some data
conn.execute('INSERT INTO my_table (id, name) VALUES (?, ?)', (1, 'John'))
conn.execute('INSERT INTO my_table (id, name) VALUES (?, ?)', (2, 'Jane'))

# Commit the changes
conn.commit()

# Retrieve data
cursor = conn.execute('SELECT * FROM my_table')
for row in cursor:
    print(row)

51. Библиотека Requests. Используйте библиотеку Requests для выполнения HTTP-запросов. Библиотека Requests предоставляет простой способ отправки HTTP-запросов к API или веб-сайтам. Вот пример того, как сделать запрос GET:

import requests

# make a GET request to a website
response = requests.get('https://www.google.com')

# print the response content
print(response.content)

52. Библиотека ОС. Используйте библиотеку ОС для управления файлами и каталогами. Библиотека ОС предоставляет функции для взаимодействия с файлами и каталогами.

import os

# create a directory
os.mkdir('my_directory')

53. Работа с JSON:

Кодирование данных Python в формат JSON:

import json

data = {
    "name": "Mark",
    "age": 28,
    "gender": "Male"
}

json_data = json.dumps(data)
print(json_data)

Декодирование данных JSON в формат Python:

import json

json_data = '{"name": "Mark", "age": 28, "gender": "Male"}'

data = json.loads(json_data)
print(data)

54. Работа с файлами CSV: Использование модуля CSV:

import csv

# Reading a CSV file
with open('example.csv', 'r') as file:
    csv_reader = csv.reader(file)
    for row in csv_reader:
        print(row)

# Writing to a CSV file
with open('example.csv', 'w', newline='') as file:
    csv_writer = csv.writer(file)
    csv_writer.writerow(['Name', 'Age', 'Gender'])
    csv_writer.writerow(['John', 25, 'Male'])
    csv_writer.writerow(['Jane', 30, 'Female'])

55. Использование SQL Alchemy для доступа к базе данных: SQL Alchemy — это популярная библиотека Python для работы с базами данных. Она предоставляет простой интерфейс для подключения к различным базам данных и выполнения запросов SQL.

from sqlalchemy import create_engine

# Connect to a PostgreSQL database
engine = create_engine('postgresql://username:password@host:port/database_name')

# Execute a SQL query and return the results as a dataframe
query = "SELECT * FROM table_name WHERE column_name > 100"
df = pd.read_sql(query, engine)

# Write a dataframe to a new table in the database
df.to_sql('new_table_name', engine)

56. Выбор признаков с использованием рекурсивного исключения признаков (RFE):

• RFE помогает определить наиболее важные функции, что приводит к повышению производительности модели и ускорению обучения.
• Выбор функций может уменьшить переоснащение и улучшить обобщение вашей модели.

from sklearn.datasets import load_iris
from sklearn.feature_selection import RFE
from sklearn.linear_model import LogisticRegression

# Load your data
iris = load_iris()
X, y = iris.data, iris.target

# Create a Logistic Regression model
model = LogisticRegression()

# Perform Recursive Feature Elimination
rfe = RFE(model, n_features_to_select=2)
rfe.fit(X, y)

# Get the most important features
important_features = rfe.support_

57. Используйте Apache Parquet для эффективного хранения данных в столбцах. Apache Parquet — это формат файла хранения в столбцах, который обеспечивает эффективные схемы сжатия и кодирования, что делает его идеальным для хранения больших наборов данных, используемых в машинном обучении.

import pandas as pd
import pyarrow as pa
import pyarrow.parquet as pq

# Read a CSV file using pandas
data = pd.read_csv("data.csv")

# Convert the pandas DataFrame to an Apache Arrow Table
table = pa.Table.from_pandas(data)

# Write the Arrow Table to a Parquet file
pq.write_table(table, "data.parquet")

# Read the Parquet file into a pandas DataFrame
data_from_parquet = pq.read_table("data.parquet").to_pandas()

58. Используйте Apache Kafka для потоковой передачи данных в реальном времени. Apache Kafka — это распределённая платформа потоковой передачи, которая позволяет создавать конвейеры данных и приложения в реальном времени.

from kafka import KafkaProducer, KafkaConsumer

# Create a Kafka producer
producer = KafkaProducer(bootstrap_servers="localhost:9092")

# Send a message to a Kafka topic
producer.send("my_topic", b"Hello, Kafka!")

# Create a Kafka consumer
consumer = KafkaConsumer("my_topic", bootstrap_servers="localhost:9092")

# Consume messages from the Kafka topic
for msg in consumer:
    print(msg.value)

59. Разделите данные для эффективного выполнения запросов. Разделение данных может помочь повысить производительность запросов за счёт уменьшения объёма данных, которые необходимо прочитать для данного запроса.

import pandas as pd
import pyarrow as pa
import pyarrow.parquet as pq

# Read a CSV file using pandas
data = pd.read_csv("data.csv")

# Convert the pandas DataFrame to an Apache Arrow Table
table = pa.Table.from_pandas(data)

# Write the Arrow Table to a partitioned Parquet dataset
pq.write_to_dataset(table, root_path="partitioned_data", partition_cols=["state"])

# Read the partitioned Parquet dataset into a pandas DataFrame
data_from_partitioned_parquet = pq.ParquetDataset("partitioned_data").read().to_pandas()

60. Используйте методы расширения данных для увеличения размера набора данных: Расширение данных включает создание новых обучающих примеров путём применения различных преобразований к существующим данным, что может помочь улучшить производительность модели.

import numpy as np
import tensorflow as tf
from tensorflow.keras.preprocessing.image import ImageDataGenerator

# Define an image data generator for data augmentation
datagen = ImageDataGenerator(
    rotation_range=20,
    width_shift_range=0.2,
    height_shift_range=0.2,
    horizontal_flip=True,
)

# Load your data
(x_train, y_train), (_, _) = tf.keras.datasets.cifar10.load_data()
x_train = x_train.astype(np.float32) / 255.0

# Fit the data generator to your data
datagen.fit(x_train)

# Train your model with augmented data
model = create_your_model()
model.compile(optimizer="adam", loss="sparse_categorical_crossentropy", metrics=["accuracy"])
model.fit(datagen.flow(x_train, y_train, batch_size=32), epochs=10)

61. Использование Flask для развёртывания модели. Ниже приведён пример использования Flask для развёртывания модели машинного обучения:

from flask import Flask, request, jsonify
import joblib

app = Flask(__name__)

@app.route('/predict', methods=['POST'])
def predict():
    data = request.get_json()
    features = [data['feature1'], data['feature2'], data['feature3']]
    model = joblib.load('model.pkl')
    prediction = model.predict([features])[0]
    response = {'prediction': int(prediction)}
    return jsonify(response)

if __name__ == '__main__':
    app.run()

62. Использование Pytest для тестирования:

Например, у нас есть файл math_operations.py.

# math_operations.py

def add(a, b):
    return a + b

def multiply(a, b):
    return a * b

Затем создайте тестовый модуль с тем же именем, что и у вашего модуля, но с префиксом test_. В нашем случае мы создадим файл с именем test_math_operations.py:

# test_math_operations.py

import math_operations

def test_add():
    assert math_operations.add(2, 3) == 5
    assert math_operations.add(-1, 1) == 0
    assert math_operations.add(0, 0) == 0

def test_multiply():
    assert math_operations.multiply(2, 3) == 6
    assert math_operations.multiply(-1, 1) == -1
    assert math_operations.multiply(0, 0) == 0

Запустите тесты с помощью команды pytest:

pytest test_math_operations.py

Pytest обнаружит и запустит тестовые функции в модуле test_math_operations.py.

63. Используйте автоматизированные конвейеры данных. Автоматизированные конвейеры данных могут помочь вам автоматизировать процесс приёма, очистки и преобразования данных. Некоторые из важных инструментов

• Apache Airflow
• Prefect
• Apache Beam
• Luigi
• Dagster
• Argo Workflows
• NiFi

Конвейер Apache Airflow Ml:

from airflow import DAG
from airflow.operators.python_operator import PythonOperator
from datetime import datetime

def preprocess_data():
    # Preprocess data here
    pass

def train_model():
    # Train model here
    pass

default_args = {
    'owner': 'myname',
    'start_date': datetime(2023, 3, 15),
    'retries': 1,
    'retry_delay': timedelta(minutes=5),
}

with DAG('my_dag', default_args=default_args, schedule_interval='@daily') as dag:
    preprocess_task = PythonOperator(task_id='preprocess_task', python_callable=preprocess_data)
    train_task = PythonOperator(task_id='train_task', python_callable=train_model)

    preprocess_task >> train_task

64. Используйте Transfer Learning: Transfer Learning может помочь вам повторно использовать и адаптировать предварительно обученные модели машинного обучения для ваших собственных вариантов использования. Вот пример того, как использовать Transfer Learning с TensorFlow:

import tensorflow as tf
from tensorflow.keras.applications import VGG16

# Load pre-trained model
base_model = VGG16(weights='imagenet', include_top=False, input_shape=(224, 224, 3))

# Freeze base layers
for layer in base_model.layers:
    layer.trainable = False

# Add custom top layers
x = base_model.output
x = tf.keras.layers.GlobalAveragePooling2D()(x)
x = tf.keras.layers.Dense(256, activation='relu')(x)
predictions = tf.keras.layers.Dense(10, activation='softmax')(x)

# Create new model
model = tf.keras.models.Model(inputs=base_model.input, outputs=predictions)

# Compile and train model
model.compile(optimizer='adam', loss='categorical_crossentropy', metrics=['accuracy'])
model.fit(X_train, y_train, epochs=10)

65. Используйте автоматизированное машинное обучение (Auto ML). Используя такие платформы, как H2O.ai или Google Cloud AutoML, вы можете автоматически выбирать, обучать и развёртывать модели на основе ваших данных и требований. Вот пример того, как использовать платформу AutoML H2O.ai:

import h2o
from h2o.automl import H2OAutoML

# Start H2O cluster
h2o.init()

# Load data
data = h2o.import_file('my-data.csv')

# Define target variable
target = 'label'

# Split data into train and test sets
train, test = data.split_frame(ratios=[0.8])

# Define AutoML settings
automl = H2OAutoML(max_models=10, seed=1234)

# Train AutoML model
automl.train(x=data.columns, y=target, training_frame=train)

# Evaluate AutoML model
predictions = automl.leader.predict(test)
accuracy = (predictions['predict'] == test[target]).mean()
print(f'Accuracy: {accuracy}')

66. Используйте обнаружение аномалий. Используя такие библиотеки, как PyOD или TensorFlow, вы можете обнаруживать аномалии на основе статистических методов или методов машинного обучения. Вот пример того, как использовать PyOD для обнаружения аномалий в наборе данных:

import numpy as np
from pyod.models.knn import KNN

# Load data
X = np.load('my-data.npy')

# Define anomaly detector
detector = KNN(n_neighbors=5)

# Train detector
detector.fit(X)

# Detect anomalies
anomaly_scores = detector.decision_scores_
threshold = np.percentile(anomaly_scores, 95)
anomalies = np.where(anomaly_scores > threshold)

# Print anomalies
print(f'Anomalies: {anomalies}')

67. Использование Weights и Biases. Вот пример использования Weights & Biases для запуска и отслеживания экспериментов по машинному обучению.

import wandb
import tensorflow as tf

# Initialize W&B
wandb.init(project='my-project')

# Load data
data = tf.data.TFRecordDataset('my-data.tfrecord')

# Define hyperparameters
config = wandb.config
config.learning_rate = 0.1
config.num_epochs = 10

# Define model
model = tf.keras.models.Sequential([
    tf.keras.layers.Dense(32, activation='relu'),
    tf.keras.layers.Dense(10, activation='softmax')
])
model.compile(optimizer=tf.keras.optimizers.Adam(), loss='categorical_crossentropy', metrics=['accuracy'])

# Train model
history = model.fit(data.batch(32), epochs=config.num_epochs)

# Log metrics and artifacts to W&B
wandb.log({'accuracy': history.history['accuracy'][-1]})
wandb.log_artifact('my-model.h5')

68. Важные инструменты управления рабочими процессами машинного обучения:

• Kubeflow
• Apache Airflow
• MLFlow
• Seldon
• Pachyderm
• DVC

69. Используйте сжатие данных. Рассмотрите возможность использования инструментов и библиотек, таких как zlib, gzip или bz2, для сжатия данных в Python.

import zlib

# Compress data with zlib
data_compressed = zlib.compress(data)

# Decompress data with zlib
data_decompressed = zlib.decompress(data_compressed)

70. Сериализация данных. Рассмотрите возможность использования инструментов и библиотек, таких как JSON, YAML или protobuf, для сериализации данных в Python.

import json

# Serialize data to JSON
data_json = json.dumps(data)

# Deserialize data from JSON
data_deserialized = json.loads(data_json) 

71. Нормализация и масштабирование данных. Рассмотрите возможность использования инструментов и библиотек, таких как scikit-learn, TensorFlow или PyTorch, для нормализации и масштабирования данных в Python.

import pandas as pd
from sklearn.preprocessing import StandardScaler, MinMaxScaler

# Standardize data with Z-score normalization
scaler = StandardScaler()
data_normalized = scaler.fit_transform(data)

# Scale data with min-max scaling
scaler = MinMaxScaler()
data_scaled = scaler.fit_transform(data)

72. Шифрование и безопасность данных. Рассмотрите возможность использования инструментов и библиотек, таких как cryptography, Fernet или PyAesCrypt, для шифрования данных и безопасности в Python.

from cryptography.fernet import Fernet

# Generate encryption key with Fernet
key = Fernet.generate_key()

# Encrypt data with Fernet
cipher_suite = Fernet(key)
encrypted_data = cipher_suite.encrypt(data)

# Decrypt data with Fernet
decrypted_data = cipher_suite.decrypt(encrypted_data)

import hashlib

# Hash data with hashlib
hash_value = hashlib.sha256(data.encode('utf-8')).hexdigest()

import tokenizers

# Define tokenization with tokenizers
tokenizer = tokenizers.Tokenizer(tokenizers.models.WordPiece('vocab.txt', unk_token='[UNK]'))
encoded_data = tokenizer.encode(data).ids

73. Проверка данных с использованием Great Expectation:

import great_expectations as ge

# Load a dataset (e.g., a Pandas DataFrame)
data = ge.read_csv("data.csv")

# Create an Expectation Suite
expectation_suite = data.create_expectation_suite("my_suite")

# Add expectations
data.expect_column_values_to_be_unique("id")
data.expect_column_values_to_not_be_null("name")
data.expect_column_mean_to_be_between("age", min_value=20, max_value=40)

# Validate data against the Expectation Suite
validation_result = data.validate(expectation_type="basic")

# Save the Expectation Suite and the validation result
ge.save_expectation_suite(expectation_suite, "my_suite.json")
ge.save_validation_result(validation_result, "my_suite_validation.json")

74. Модуль logging : используйте модуль logging  для гибкого логирования.

import logging

logging.basicConfig(level=logging.INFO)
logging.info("This is an info message.")
logging.error("This is an error message.")

75. Используйте фрейм данных Dask: Dask — мощная библиотека для параллельных и распределённых вычислений на Python. Она позволяет обрабатывать большие наборы данных, которые не помещаются в память, разбивая их на более мелкие фрагменты и обрабатывая их параллельно.

import dask.dataframe as dd

# Read CSV file using Dask (file is partitioned into smaller chunks)
ddf = dd.read_csv('large_file.csv')

# Perform operations on the data (lazy evaluation)
filtered_ddf = ddf[ddf['column_A'] > 10]
mean_value = filtered_ddf['column_B'].mean()

# Compute the result (operations are executed in parallel)
result = mean_value.compute()
print("Mean of column B for rows where column A > 10:", result)

Заключение

Я надеюсь, что приведённые выше советы и рекомендации будут полезны для вас. Опять же, в MLOps много инструментов и процессов. Ландшафт MLOps также продолжает меняться очень быстро. Я советую вам постоянно следить за актуальными обновлениями в данной сферы, если вы хотите продвигаться в ней!

Список используемый литературы:

1. Mlflow: An Open Platform to Simplify the Machine Learning Lifecycle. (2021). https://mlflow.org/
2. Prometheus: An open-source monitoring system with a dimensional data model. (2021). https://prometheus.io/
3. Grafana: The open and composable observability and data visualization platform. (2021). https://grafana.com/
4. Seldon Core: Deploy, scale & monitor your machine learning models in Kubernetes. (2021).https://www.seldon.io/tech/products/core/
5. Kubeflow: The Machine Learning Toolkit for Kubernetes. (2021). https://www.kubeflow.org/
6. Dask: Parallel computing with task scheduling. (2021). Retrieved from https://dask.org/
7. Great Expectations: Always know what to expect from your data. (2021). https://greatexpectations.io/
8. Airflow: A platform to programmatically author, schedule, and monitor workflows. (2021). https://airflow.apache.org/
9. TensorFlow Extended: A production-ready ML platform for TensorFlow. (2021). https://www.tensorflow.org/tfx
10. Prefect: The New Standard in Dataflow Automation. (2021). https://www.prefect.io/
11. Feast: Feature Store for Machine Learning. (2021). https://feast.dev/
12. “Machine Learning Engineering” by Andriy Burkov.
13. “Data Engineering Cookbook” by Andreas Kretz.
14. “Hands-On Data Engineering with Python” by James Lee.