ML Migration — Databricks to Fabric + Azure ML¶

# Databricks notebook
import mlflow
import mlflow.sklearn
from sklearn.ensemble import RandomForestClassifier
from sklearn.metrics import accuracy_score

mlflow.set_experiment("/Users/user@company.com/customer-churn")

with mlflow.start_run(run_name="rf-baseline"):
    # Train model
    model = RandomForestClassifier(n_estimators=100, max_depth=10)
    model.fit(X_train, y_train)
    predictions = model.predict(X_test)

    # Log parameters, metrics, model
    mlflow.log_param("n_estimators", 100)
    mlflow.log_param("max_depth", 10)
    mlflow.log_metric("accuracy", accuracy_score(y_test, predictions))
    mlflow.log_metric("f1_score", f1_score(y_test, predictions))
    mlflow.sklearn.log_model(model, "model")

    # Register model in Unity Catalog
    mlflow.register_model("runs:/{run_id}/model", "catalog.schema.churn_model")

# Fabric notebook
import mlflow
import mlflow.sklearn
from sklearn.ensemble import RandomForestClassifier
from sklearn.metrics import accuracy_score

# Fabric automatically configures the MLflow tracking URI
# No need to set_tracking_uri or set_experiment manually

mlflow.set_experiment("customer-churn")

with mlflow.start_run(run_name="rf-baseline"):
    # Train model -- same code as Databricks
    model = RandomForestClassifier(n_estimators=100, max_depth=10)
    model.fit(X_train, y_train)
    predictions = model.predict(X_test)

    # Log parameters, metrics, model -- same API
    mlflow.log_param("n_estimators", 100)
    mlflow.log_param("max_depth", 10)
    mlflow.log_metric("accuracy", accuracy_score(y_test, predictions))
    mlflow.log_metric("f1_score", f1_score(y_test, predictions))
    mlflow.sklearn.log_model(model, "model")

    # Save model to Fabric ML model registry
    mlflow.register_model("runs:/{run_id}/model", "churn_model")

# Databricks: Deploy model to serving endpoint
from databricks.sdk import WorkspaceClient

w = WorkspaceClient()
w.serving_endpoints.create(
    name="churn-predictor",
    config={
        "served_models": [{
            "model_name": "churn_model",
            "model_version": "3",
            "workload_size": "Small",
            "scale_to_zero_enabled": True,
        }]
    }
)

# Azure ML: Deploy model from Fabric ML registry
from azure.ai.ml import MLClient
from azure.ai.ml.entities import (
    ManagedOnlineEndpoint,
    ManagedOnlineDeployment,
    Model,
    Environment,
)
from azure.identity import DefaultAzureCredential

ml_client = MLClient(
    DefaultAzureCredential(),
    subscription_id="<sub-id>",
    resource_group_name="<rg>",
    workspace_name="<aml-workspace>"
)

# Register model (or reference from Fabric OneLake)
model = Model(
    path="azureml://datastores/onelake/paths/models/churn/model.pkl",
    name="churn-model",
    type="custom_model"
)
ml_client.models.create_or_update(model)

# Create endpoint
endpoint = ManagedOnlineEndpoint(name="churn-predictor", auth_mode="key")
ml_client.online_endpoints.begin_create_or_update(endpoint).result()

# Create deployment
deployment = ManagedOnlineDeployment(
    name="v1",
    endpoint_name="churn-predictor",
    model="churn-model:1",
    instance_type="Standard_DS3_v2",
    instance_count=1,
)
ml_client.online_deployments.begin_create_or_update(deployment).result()

# Databricks: Create and publish features
from databricks.feature_engineering import FeatureEngineeringClient

fe = FeatureEngineeringClient()

# Create feature table
fe.create_table(
    name="production.features.customer_features",
    primary_keys=["customer_id"],
    timestamp_keys=["feature_timestamp"],
    df=customer_features_df,
    description="Customer behavioral features for churn prediction"
)

# Use features for training
training_set = fe.create_training_set(
    df=labels_df,
    feature_lookups=[
        FeatureLookup(
            table_name="production.features.customer_features",
            lookup_key="customer_id"
        )
    ],
    label="churned"
)
training_df = training_set.load_df()

# Fabric: Feature engineering
from microsoft.fabric.ml.feature_store import (
    FeatureStoreClient,
    FeatureSet,
)

fs_client = FeatureStoreClient()

# Define feature set
feature_set = FeatureSet(
    name="customer_features",
    entities=[{"name": "customer_id", "key": True}],
    features=customer_features_df,
    description="Customer behavioral features"
)
fs_client.register_feature_set(feature_set)

# Retrieve features for training
training_df = fs_client.get_features(
    feature_set_name="customer_features",
    entity_df=labels_df
)

# Databricks AutoML
from databricks import automl

summary = automl.classify(
    dataset=training_df,
    target_col="churned",
    primary_metric="f1",
    timeout_minutes=30,
    max_trials=50
)

# Best model is logged to MLflow
best_run = summary.best_trial
print(f"Best F1: {best_run.metrics['f1_score']}")

# Fabric AutoML
import flaml

automl_settings = {
    "task": "classification",
    "metric": "f1",
    "time_budget": 1800,  # 30 minutes
    "log_file_name": "automl.log",
}

automl_model = flaml.AutoML()
automl_model.fit(X_train, y_train, **automl_settings)

# Log to Fabric ML experiments
import mlflow
with mlflow.start_run():
    mlflow.log_params(automl_model.best_config)
    mlflow.log_metric("f1", automl_model.best_loss)
    mlflow.sklearn.log_model(automl_model.model, "model")

# Databricks: Create vector search index
from databricks.vector_search.client import VectorSearchClient

vsc = VectorSearchClient()

# Create endpoint
vsc.create_endpoint(name="vs-endpoint", endpoint_type="STANDARD")

# Create index on Delta table with embeddings
vsc.create_delta_sync_index(
    endpoint_name="vs-endpoint",
    index_name="catalog.schema.doc_embeddings_index",
    source_table_name="catalog.schema.documents",
    pipeline_type="TRIGGERED",
    primary_key="doc_id",
    embedding_source_column="content",
    embedding_model_endpoint_name="databricks-bge-large-en"
)

# Search
results = vsc.get_index("catalog.schema.doc_embeddings_index").similarity_search(
    query_text="How to reset password",
    columns=["doc_id", "title", "content"],
    num_results=5
)

# Azure AI Search: Vector search
from azure.search.documents import SearchClient
from azure.search.documents.indexes import SearchIndexClient
from azure.search.documents.indexes.models import (
    SearchIndex,
    SearchField,
    VectorSearch,
    HnswAlgorithmConfiguration,
    VectorSearchProfile,
    SearchFieldDataType,
)
from azure.identity import DefaultAzureCredential

# Create index with vector field
index = SearchIndex(
    name="documents",
    fields=[
        SearchField(name="doc_id", type=SearchFieldDataType.String, key=True),
        SearchField(name="title", type=SearchFieldDataType.String, searchable=True),
        SearchField(name="content", type=SearchFieldDataType.String, searchable=True),
        SearchField(
            name="content_vector",
            type=SearchFieldDataType.Collection(SearchFieldDataType.Single),
            vector_search_dimensions=1536,
            vector_search_profile_name="my-profile"
        ),
    ],
    vector_search=VectorSearch(
        algorithms=[HnswAlgorithmConfiguration(name="hnsw")],
        profiles=[VectorSearchProfile(name="my-profile", algorithm_configuration_name="hnsw")]
    )
)

# Search
from azure.search.documents.models import VectorizedQuery

search_client = SearchClient(endpoint, "documents", credential)
results = search_client.search(
    search_text="How to reset password",
    vector_queries=[VectorizedQuery(
        vector=embedding_vector,
        k_nearest_neighbors=5,
        fields="content_vector"
    )]
)