MLOps-Советы и хитрости-75 фрагментов кода

MLOps-Советы и хитрости-75 фрагментов кода

MLOps и обработка данных

Введение:

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:

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

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

# Install TensorFlow Serving
echo "deb [arch=amd64] https://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), чтобы объяснить выходные данные ваших моделей машинного обучения и получить представление об их поведении.

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:конвейер: используйте 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: поиск по сетке: используйте 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. Кодировщики категорий.Кодировщики категорий — это библиотека, предоставляющая различные методы кодирования для категориальных переменных, такие как целевое кодирование, однократное кодирование и порядковое кодирование.

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:ColumnTransformer позволяет применять различные этапы предварительной обработки к разным столбцам входных данных, что особенно полезно при работе со смешанными типами данных.

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. Используйте проверку данных TensorFlow (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. Используйте анализ моделей TensorFlow (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: используйте преобразование TensorFlow (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, для вычислений с ускорением на графическом процессоре. Используйте массивы 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: создайте файл 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:

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 позволяет параметризовать Jupyter 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 для интеграции пользовательских функций или операций в рабочий процесс цепочки 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. Библиотека запросов.Используйте библиотеку запросов для выполнения HTTP-запросов. Библиотека запросов предоставляет простой способ отправки 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. Библиотека ОС:Используйте библиотеку os для управления файлами и каталогами. Библиотека os предоставляет функции для взаимодействия с файлами и каталогами.

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 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. Используйте трансферное обучение. Трансферное обучение может помочь вам повторно использовать и адаптировать предварительно обученные модели машинного обучения для ваших собственных сценариев использования. Вот пример того, как использовать трансферное обучение с 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. Использование весов и смещений. Вот пример использования весов и смещений для запуска и отслеживания экспериментов по машинному обучению.

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. Важные инструменты для управления рабочими процессами машинного обучения:

• Кубфлоу
• Апач Эйрфлоу
• МЛФлоу
• Селдон
• Толстокожий
• ДВК

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. Шифрование и безопасность данных. Рассмотрите возможность использования инструментов и библиотек, таких как криптография, 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. Проверка данных с использованием больших ожиданий:

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 также продолжает меняться очень быстро. Лучше постоянно обновлять последние инструменты и процессы MLOps.

Использованная литература:

1. Mlflow: открытая платформа для упрощения жизненного цикла машинного обучения. (2021). https://mlflow.org/
2. Prometheus: система мониторинга с открытым исходным кодом и многомерной моделью данных. (2021). https://prometheus.io/
3. Grafana: открытая и компонуемая платформа для наблюдения и визуализации данных. (2021). https://grafana.com/
4. Seldon Core: развертывание, масштабирование и мониторинг моделей машинного обучения в Kubernetes. (2021). https://www.seldon.io/tech/products/core/
5. Kubeflow: набор инструментов машинного обучения для Kubernetes. (2021). https://www.kubeflow.org/
6. Dask: параллельные вычисления с планированием задач. (2021). Получено с https://dask.org/
7. Большие ожидания: всегда знайте, чего ожидать от ваших данных. (2021). https://greateexpectations.io/
8. Airflow: платформа для программного создания, планирования и мониторинга рабочих процессов. (2021). https://airflow.apache.org/
9. TensorFlow Extended: готовая к производству платформа машинного обучения для TensorFlow. (2021). https://www.tensorflow.org/tfx
10. Префект: новый стандарт автоматизации потоков данных. (2021). https://www.prefect.io/
11. Feast: магазин функций для машинного обучения. (2021). https://feast.dev/
12. «Инженерия машинного обучения» Андрея Буркова.
13. «Поваренная книга по инженерии данных» Андреаса Кретца.
14. «Практическая обработка данных с помощью Python» Джеймса Ли.