با Databricks و MLflow یک خط لوله ML سرتاسر ایجاد کنید

در سازمان‌های مبتنی بر داده، ایجاد یک خط لوله یادگیری ماشینی (ML) که تکرارپذیر، مقیاس‌پذیر و قابل ردیابی باشد، یک جزء ضروری است. اکوسیستم یکپارچه Delta Lake، Auto Loader و MLflow در Databricks، سازمان‌ها را قادر می‌سازد تا چرخه حیات ML را از دریافت داده‌های خام تا استقرار تولید ساده‌سازی کنند.

این آموزش یک راهنمای جامع برای ساخت یک خط لوله ML سرتاسر بر روی Databricks، استفاده از MLflow برای ردیابی مدل و ثبت مدل و استفاده از Delta Lake برای مدیریت داده ها ارائه می دهد. ما همه وظایف را در یک گردش کاری یکپارچه نشان خواهیم داد، از جمله دریافت داده های خام، آماده سازی ویژگی، آموزش مدل، و خدمات پیش بینی.

نمای کلی معماری

برنز → نقره → طلا → MLflow → رجیستری مدل → ارزیابی دسته ای.

خط لوله ما بر اساس چارچوب کلاسیک برنز-نقره-طلا ساخته شده است.

• برنز: مرحله دریافت داده های خام با استفاده از بارگذار خودکار.
• نقره ای: داده های پاک شده که برای تجزیه و تحلیل قابل اعتماد کپی شده اند.
• طلا: ویژگی های طراحی شده ای که برای آموزش و ارزیابی مدل آماده می شود.
• ML: آموزش و پیگیری مدل با ثبت نام و خروجی دسته ای.

1. ذخیره سازی داده ها در لایه برنز

اولین کار جمع آوری داده های خام به کارآمدترین و افزایشی ترین روش ممکن است. با توجه به این کار، Databricks Auto Loader می تواند فرآیند را ساده کند زیرا به طور خودکار فایل های جدید را در پوشه ابری مشخص شده ردیابی می کند و آنها را در جداول دلتا بارگذاری می کند.

catalog = "main"
schema  = "ml_demo"
raw_dir = "dbfs:/mnt/raw/churn/"
bronze  = f"{catalog}.{schema}.bronze_churn"

spark.sql(f"CREATE SCHEMA IF NOT EXISTS {catalog}.{schema}")

from pyspark.sql.types import *
schema_hint = StructType([
    StructField("customer_id", StringType()),
    StructField("tenure", IntegerType()),
    StructField("monthly_charges", DoubleType()),
    StructField("contract_type", StringType()),
    StructField("churn", IntegerType())
])

فایل ها را مستقیماً با autoloader آپلود کنید.

spark.readStream
 .format("cloudFiles")
 .option("cloudFiles.format", "csv")
 .option("header", "true")
 .schema(schema_hint)
 .load(raw_dir)
 .writeStream
 .trigger(availableNow=True)
 .option("checkpointLocation", "dbfs:/chk/bronze_churn")
 .toTable(bronze)

2. پاکسازی و تبدیل داده ها (لایه نقره ای)

پس از فرود در ردیف برنز، ما به تمیز کردن و استاندارد کردن داده ها به طور مستقیم در جدول نقره ای اقدام می کنیم.

silver = f"{catalog}.{schema}.silver_churn"

df_bronze = spark.table(bronze)

from pyspark.sql.functions import col, trim
df_silver = (df_bronze
  .dropDuplicates(["customer_id"])
  .withColumn("contract_type", trim(col("contract_type")))
  .filter(col("tenure").isNotNull() & col("monthly_charges").isNotNull()))

df_silver.write.mode("overwrite").format("delta").saveAsTable(silver)

بررسی کیفیت داده ها (بهترین عمل): در این مرحله از بررسی هایی استفاده کنید، از جمله بررسی های نسبت تهی، اعتبار سنجی طرحواره و معیارهای تکراری.

3. مهندسی ویژگی (لایه طلا)

در لایه طلا، ما تغییراتی را روی داده های پاک شده انجام می دهیم تا ویژگی هایی ایجاد کنیم که در مدل های یادگیری ماشین استفاده می شود.

gold = f"{catalog}.{schema}.gold_churn_features"

from pyspark.sql.functions import when

df_gold = (spark.table(silver)
  .withColumn("is_long_tenure", (col("tenure") >= 12).cast("int"))
  .withColumn("contract_monthly", (col("contract_type")=="Monthly").cast("int"))
  .select("customer_id","tenure","monthly_charges","is_long_tenure","contract_monthly","churn"))

df_gold.write.mode("overwrite").format("delta").saveAsTable(gold)

نتیجه یک میز طلایی است که برای آموزش مدل آماده استفاده است.

4. آموزش و ردیابی مدل ها با استفاده از MLflow

اکنون به MLflow می‌رویم، جایی که مدل‌ها را اجرا می‌کنیم و آزمایش‌ها را دنبال می‌کنیم.

import mlflow
import mlflow.sklearn
from sklearn.model_selection import train_test_split
from sklearn.ensemble import RandomForestClassifier
from sklearn.metrics import f1_score, roc_auc_score
import pandas as pd

mlflow.set_experiment("/Shared/churn_ml_experiment")

pdf = spark.table(gold).toPandas()
X = pdf.drop(columns=["customer_id","churn"])
y = pdf["churn"]

X_train, X_test, y_train, y_test = train_test_split(X, y, test_size=0.2, random_state=42, stratify=y)

with mlflow.start_run():
    params = {"n_estimators": 200, "max_depth": 6, "random_state": 42}
    clf = RandomForestClassifier(**params)
    clf.fit(X_train, y_train)

    preds = clf.predict(X_test)
    probas = clf.predict_proba(X_test)[:,1]

    mlflow.log_params(params)
    mlflow.log_metrics({
        "f1": f1_score(y_test, preds),
        "roc_auc": roc_auc_score(y_test, probas)
    })
    mlflow.sklearn.log_model(clf, "model", registered_model_name="churn_rf_model")

موارد زیر برخی از مواردی است که MLflow ادغام می کند:

• پارامترها (به عنوان مثال تعداد درختان)
• معیارها (به عنوان مثال امتیاز F1، ROC AUC)
• مصنوعات (فایل های مدل، گرافیک)

هنگامی که یک مدل عملکرد بهینه داشت، می‌توانیم آن را برای ارتقای عمده در مراحل چرخه حیات ثبت کنیم.

5. الگوهای ثبت و انتقال

from mlflow.tracking import MlflowClient
client = MlflowClient()
name = "churn_rf_model"

latest = client.get_latest_versions(name, stages=["None"])[-1]
client.transition_model_version_stage(
    name=name,
    version=latest.version,
    stage="Staging"
)

نکته: تبلیغات را با Databricks Jobs و گردش کار تایید سفارشی به صورت خودکار انجام دهید.

6. نتیجه گیری دسته ای برای به ثمر رساندن داده های جدید

اکنون می‌توانیم مدل را برای تخمین احتمال ریزش مشتری برای داده‌های جدید بارگذاری کنیم.

این اجازه می دهد تا داده ها در یک جدول پیش بینی ذخیره شوند که می تواند برای ایجاد داشبورد مشتری و سیستم های نگهداری استفاده شود.

prod_uri = "models:/churn_rf_model/Production"
import mlflow.pyfunc
model = mlflow.pyfunc.load_model(prod_uri)

unscored = (spark.table(silver)
  .select("customer_id","tenure","monthly_charges","contract_type")
  .withColumn("is_long_tenure", (col("tenure") >= 12).cast("int"))
  .withColumn("contract_monthly", (col("contract_type")=="Monthly").cast("int"))
  .drop("contract_type"))

pdf_unscored = unscored.toPandas()
scores = model.predict_proba(pdf_unscored.drop(columns=["customer_id"]))[:,1]

import pandas as pd
out = pd.DataFrame({
    "customer_id": pdf_unscored["customer_id"],
    "churn_score": scores
})

spark.createDataFrame(out).write.mode("overwrite").saveAsTable(f"{catalog}.{schema}.predictions_churn")

7. خودکار و نظارت

در این جنبه، Databricks با هدف ساده سازی فرآیند تولید از طریق:

• مشاغل: جذب و آموزش مدل با اتصال سه فرآیند در یک گردش کار برنامه ریزی شده به دست می آید.
• مخازن: نسخه های نوت بوک را با استفاده از Git کنترل کنید.
• جداول زنده دلتا (DLT): کیفیت داده ها و بررسی های منشأ را خودکار کنید.
• نظارت بر مدل: رانش و تأخیر را اندازه‌گیری کنید و دقت را با استفاده از معیارهای MLflow اندازه‌گیری کنید.
• بهترین تمرین: برای مقایسه عملکرد تولید، کار بازآموزی شبانه را برای هر مدل جدید برنامه ریزی کنید.

8. بهینه سازی هزینه ها و بهره وری

• برای بلع کارآمدتر، Auto Loader را با فهرست افزایشی ترکیب کنید.
• از OPTIMIZE با ZORDER برای فشرده سازی فایل های کوچک استفاده کنید.
• بارگیری های ML را با Hyperopt یا RandomizedSearchCV تنظیم کنید.
• برای کارهای غیر حیاتی، از خوشه های نقطه ای یا قابل جابجایی استفاده کنید.
• مهندسی آگاهانه هزینه تضمین می کند که خطوط لوله ML شما مقیاس پذیر و انعطاف پذیر هستند.

نتیجه گیری

ترکیب Databricks Delta Lake و عملکرد مدیریت داده قوی آن، همراه با MLflow برای ردیابی آزمایشی، عملکرد یکپارچه خطوط لوله ML را امکان پذیر می کند.

در این آموزش، من نشان دادم که چگونه با استفاده از معماری برنز-نقره-طلا، یک گردش کار سرتاسر قابل تکرار – از پذیرش تا امتیازدهی – بسازیم. می‌توانید با ماژول‌های Feature Store، امتیازدهی هم‌زمان، یا ماژول‌های اتوماسیون MLOps برای تحویل مداوم، موارد بیشتری اضافه کنید.

به طور خلاصه، Databricks در حال تغییر شکل خطوط لوله ML از لپ تاپ های دستی به سیستم های مدیریت شده، خودکار و مقیاس پذیر است.