Použití funkcí k trénování modelů

  • Článek

Tento článek popisuje, jak můžete trénovat modely pomocí přípravy funkcí v katalogu Unity nebo místního úložiště funkcí pracovního prostoru. Nejprve musíte vytvořit trénovací datovou sadu, která definuje funkce, které se mají použít a jak se k nim připojit. Když pak model vytrénujete, zachová model odkazy na funkce.

Při trénování modelu pomocí přípravy funkcí v katalogu Unity můžete zobrazit rodokmen modelu v Průzkumníku katalogu. Tabulky a funkce použité k vytvoření modelu se automaticky sledují a zobrazují. Viz Zobrazení rodokmenu úložiště funkcí.

Když model použijete k odvozování, můžete zvolit, aby načetl hodnoty funkcí z úložiště funkcí. Model můžete také obsluhovat pomocí služby Model Serving a automaticky vyhledá funkce publikované v online obchodech. Modely úložiště funkcí jsou také kompatibilní s rozhraním Pyfunc MLflow, takže můžete pomocí MLflow provádět dávkové odvozování s tabulkami funkcí.

Pokud váš model používá proměnné prostředí, přečtěte si další informace o tom, jak je použít při online poskytování modelu při konfiguraci přístupu k prostředkům z koncových bodů obsluhy modelu.

Vytvoření trénovací datové sady

Pokud chcete vybrat konkrétní funkce z tabulky funkcí pro trénování modelu, vytvoříte trénovací datovou sadu pomocí FeatureEngineeringClient.create_training_set rozhraní API (pro přípravu funkcí v katalogu Unity) nebo FeatureStoreClient.create_training_set (pro úložiště funkcí pracovního prostoru) a objektu označovaného jako FeatureLookup. Určuje FeatureLookup každou funkci, která se má použít v trénovací sadě, včetně názvu tabulky funkcí, názvů funkcí a klíčů, které se mají použít při připojování tabulky funkcí s datovým rámcem předaným create_training_set. Další informace najdete v tématu Vyhledávání funkcí.

feature_names Při vytváření parametru FeatureLookuppoužijte parametr . feature_names vezme jeden název funkce, seznam názvů funkcí nebo Žádné, aby vyhledaly všechny funkce (s výjimkou primárních klíčů) v tabulce funkcí v okamžiku vytvoření trénovací sady.

Poznámka:

Typ a pořadí lookup_key sloupců v datovém rámci musí odpovídat typu a pořadí primárních klíčů (s výjimkou klíčů časového razítka) tabulky referenčních funkcí.

Tento článek obsahuje příklady kódu pro obě verze syntaxe.

V tomto příkladu obsahuje datový rámec vrácený trainingSet.load_df sloupcem pro každou funkci v objektu feature_lookups. Zachovává všechny sloupce datového rámce poskytnutého s create_training_set výjimkou těch, které jsou vyloučeny pomocí exclude_columns.

Příprava funkcí v katalogu Unity

from databricks.feature_engineering import FeatureEngineeringClient, FeatureLookup

# The model training uses two features from the 'customer_features' feature table and
# a single feature from 'product_features'
feature_lookups = [
    FeatureLookup(
      table_name='ml.recommender_system.customer_features',
      feature_names=['total_purchases_30d', 'total_purchases_7d'],
      lookup_key='customer_id'
    ),
    FeatureLookup(
      table_name='ml.recommender_system.product_features',
      feature_names=['category'],
      lookup_key='product_id'
    )
  ]

fe = FeatureEngineeringClient()

# Create a training set using training DataFrame and features from Feature Store
# The training DataFrame must contain all lookup keys from the set of feature lookups,
# in this case 'customer_id' and 'product_id'. It must also contain all labels used
# for training, in this case 'rating'.
training_set = fe.create_training_set(
  df=training_df,
  feature_lookups=feature_lookups,
  label='rating',
  exclude_columns=['customer_id', 'product_id']
)

training_df = training_set.load_df()

Úložiště funkcí pracovního prostoru

from databricks.feature_store import FeatureLookup, FeatureStoreClient

# The model training uses two features from the 'customer_features' feature table and
# a single feature from 'product_features'
feature_lookups = [
    FeatureLookup(
      table_name='recommender_system.customer_features',
      feature_names=['total_purchases_30d', 'total_purchases_7d'],
      lookup_key='customer_id'
    ),
    FeatureLookup(
      table_name='recommender_system.product_features',
      feature_names=['category'],
      lookup_key='product_id'
    )
  ]

fs = FeatureStoreClient()

# Create a training set using training DataFrame and features from Feature Store
# The training DataFrame must contain all lookup keys from the set of feature lookups,
# in this case 'customer_id' and 'product_id'. It must also contain all labels used
# for training, in this case 'rating'.
training_set = fs.create_training_set(
  df=training_df,
  feature_lookups=feature_lookups,
  label='rating',
  exclude_columns=['customer_id', 'product_id']
)

training_df = training_set.load_df()

Vytvoření trénovací sady, když vyhledávací klíče neodpovídají primárním klíčům

Použijte argument lookup_key v FeatureLookup názvu sloupce v trénovací sadě. create_training_set provede seřazené spojení mezi sloupci z trénovací sady zadané v argumentu lookup_key pomocí pořadí, ve kterém byly při vytvoření tabulky funkcí zadány primární klíče.

V tomto příkladu recommender_system.customer_features má následující primární klíče: customer_id, dt.

Tabulka recommender_system.product_features funkcí má primární klíč product_id.

training_df Pokud má následující sloupce:

  • cid
  • transaction_dt
  • product_id
  • rating

Následující kód vytvoří správné vyhledávání funkcí pro TrainingSet:

Příprava funkcí v katalogu Unity

feature_lookups = [
    FeatureLookup(
      table_name='ml.recommender_system.customer_features',
      feature_names=['total_purchases_30d', 'total_purchases_7d'],
      lookup_key=['cid', 'transaction_dt']
    ),
    FeatureLookup(
      table_name='ml.recommender_system.product_features',
      feature_names=['category'],
      lookup_key='product_id'
    )
  ]

Úložiště funkcí pracovního prostoru

feature_lookups = [
    FeatureLookup(
      table_name='recommender_system.customer_features',
      feature_names=['total_purchases_30d', 'total_purchases_7d'],
      lookup_key=['cid', 'transaction_dt']
    ),
    FeatureLookup(
      table_name='recommender_system.product_features',
      feature_names=['category'],
      lookup_key='product_id'
    )
  ]

Při create_training_set volání vytvoří trénovací datovou sadu provedením levého spojení, spojením tabulek recommender_system.customer_features a training_df použitím klíčů (,dt) odpovídajících (customer_idcid,transaction_dt), jak je znázorněno v následujícím kódu:

Příprava funkcí v katalogu Unity

customer_features_df = spark.sql("SELECT * FROM ml.recommender_system.customer_features")
product_features_df = spark.sql("SELECT * FROM ml.recommender_system.product_features")

training_df.join(
  customer_features_df,
  on=[training_df.cid == customer_features_df.customer_id,
      training_df.transaction_dt == customer_features_df.dt],
  how="left"
).join(
  product_features_df,
  on="product_id",
  how="left"
)

Úložiště funkcí pracovního prostoru

customer_features_df = spark.sql("SELECT * FROM recommender_system.customer_features")
product_features_df = spark.sql("SELECT * FROM recommender_system.product_features")

training_df.join(
  customer_features_df,
  on=[training_df.cid == customer_features_df.customer_id,
      training_df.transaction_dt == customer_features_df.dt],
  how="left"
).join(
  product_features_df,
  on="product_id",
  how="left"
)

Vytvoření trénovací sady obsahující dvě funkce se stejným názvem z různých tabulek funkcí

Použijte volitelný argument output_name v argumentu FeatureLookup. Zadaný název se používá místo názvu funkce v datovém rámci vráceného TrainingSet.load_dffunkcí . Například s následujícím kódem datový rámec vrácený training_set.load_df pomocí sloupců customer_height a product_height.

Příprava funkcí v katalogu Unity

feature_lookups = [
    FeatureLookup(
      table_name='ml.recommender_system.customer_features',
      feature_names=['height'],
      lookup_key='customer_id',
      output_name='customer_height',
    ),
    FeatureLookup(
      table_name='ml.recommender_system.product_features',
      feature_names=['height'],
      lookup_key='product_id',
      output_name='product_height'
    ),
  ]

fe = FeatureEngineeringClient()

with mlflow.start_run():
  training_set = fe.create_training_set(
    df=df,
    feature_lookups=feature_lookups,
    label='rating',
    exclude_columns=['customer_id']
  )
  training_df = training_set.load_df()

Úložiště funkcí pracovního prostoru

feature_lookups = [
    FeatureLookup(
      table_name='recommender_system.customer_features',
      feature_names=['height'],
      lookup_key='customer_id',
      output_name='customer_height',
    ),
    FeatureLookup(
      table_name='recommender_system.product_features',
      feature_names=['height'],
      lookup_key='product_id',
      output_name='product_height'
    ),
  ]

fs = FeatureStoreClient()

with mlflow.start_run():
  training_set = fs.create_training_set(
    df=df,
    feature_lookups=feature_lookups,
    label='rating',
    exclude_columns=['customer_id']
  )
  training_df = training_set.load_df()

Vytvoření trénovací sady pomocí stejné funkce několikrát

Pokud chcete vytvořit trénovací sadu pomocí stejné funkce spojené různými vyhledávacími klíči, použijte více funkcí FeatureLookups. Pro každý výstup funkce FeatureLookup použijte jedinečnou output_name možnost.

Příprava funkcí v katalogu Unity

feature_lookups = [
    FeatureLookup(
      table_name='ml.taxi_data.zip_features',
      feature_names=['temperature'],
      lookup_key=['pickup_zip'],
      output_name='pickup_temp'
    ),
    FeatureLookup(
      table_name='ml.taxi_data.zip_features',
      feature_names=['temperature'],
      lookup_key=['dropoff_zip'],
      output_name='dropoff_temp'
    )
  ]

Úložiště funkcí pracovního prostoru

feature_lookups = [
    FeatureLookup(
      table_name='taxi_data.zip_features',
      feature_names=['temperature'],
      lookup_key=['pickup_zip'],
      output_name='pickup_temp'
    ),
    FeatureLookup(
      table_name='taxi_data.zip_features',
      feature_names=['temperature'],
      lookup_key=['dropoff_zip'],
      output_name='dropoff_temp'
    )
  ]

Vytvoření trénovací sady pro modely strojového učení bez supervize

Nastavte label=None při vytváření trénovací sady pro modely učení bez dohledu. Například následující trénovací sada se dá použít ke clusteru různých zákazníků do skupin na základě jejich zájmů:

Příprava funkcí v katalogu Unity

feature_lookups = [
    FeatureLookup(
      table_name='ml.recommender_system.customer_features',
      feature_names=['interests'],
      lookup_key='customer_id',
    ),
  ]

fe = FeatureEngineeringClient()
with mlflow.start_run():
  training_set = fe.create_training_set(
    df=df,
    feature_lookups=feature_lookups,
    label=None,
    exclude_columns=['customer_id']
  )

  training_df = training_set.load_df()

Úložiště funkcí pracovního prostoru

feature_lookups = [
    FeatureLookup(
      table_name='recommender_system.customer_features',
      feature_names=['interests'],
      lookup_key='customer_id',
    ),
  ]

fs = FeatureStoreClient()
with mlflow.start_run():
  training_set = fs.create_training_set(
    df=df,
    feature_lookups=feature_lookups,
    label=None,
    exclude_columns=['customer_id']
  )

  training_df = training_set.load_df()

Trénování modelů a dávkové odvozování pomocí tabulek funkcí

Při trénování modelu pomocí funkcí z úložiště funkcí zachová model odkazy na funkce. Když model použijete k odvozování, můžete zvolit, aby načetl hodnoty funkcí z úložiště funkcí. Musíte zadat primární klíče funkcí použitých v modelu. Model načte funkce, které vyžaduje, z úložiště funkcí ve vašem pracovním prostoru. Následně spojí hodnoty funkcí podle potřeby během bodování.

Podpora vyhledávání funkcí v době odvození:

  • Model je nutné protokolovat pomocí log_model metody FeatureEngineeringClient (pro přípravu funkcí v katalogu Unity) nebo FeatureStoreClient (pro úložiště funkcí pracovního prostoru).
  • K trénování modelu je nutné použít datový rámec vrácený TrainingSet.load_df modelem. Pokud tento datový rámec upravíte jakýmkoli způsobem před jeho použitím k trénování modelu, nebudou změny použity při použití modelu pro odvozování. Tím se sníží výkon modelu.
  • Typ modelu musí mít odpovídající python_flavor typ v MLflow. MLflow podporuje většinu trénovacích architektur modelu Pythonu, včetně následujících:
    • scikit-learn
    • keras
    • PyTorch
    • SparkML
    • LightGBM
    • XGBoost
    • TensorFlow Keras (pomocí python_flavormlflow.keras)
  • Vlastní modely Pyfunc MLflow

Příprava funkcí v katalogu Unity

# Train model
import mlflow
from sklearn import linear_model

feature_lookups = [
    FeatureLookup(
      table_name='ml.recommender_system.customer_features',
      feature_names=['total_purchases_30d'],
      lookup_key='customer_id',
    ),
    FeatureLookup(
      table_name='ml.recommender_system.product_features',
      feature_names=['category'],
      lookup_key='product_id'
    )
  ]

fe = FeatureEngineeringClient()

with mlflow.start_run():

  # df has columns ['customer_id', 'product_id', 'rating']
  training_set = fe.create_training_set(
    df=df,
    feature_lookups=feature_lookups,
    label='rating',
    exclude_columns=['customer_id', 'product_id']
  )

  training_df = training_set.load_df().toPandas()

  # "training_df" columns ['total_purchases_30d', 'category', 'rating']
  X_train = training_df.drop(['rating'], axis=1)
  y_train = training_df.rating

  model = linear_model.LinearRegression().fit(X_train, y_train)

  fe.log_model(
    model=model,
    artifact_path="recommendation_model",
    flavor=mlflow.sklearn,
    training_set=training_set,
    registered_model_name="recommendation_model"
  )

# Batch inference

# If the model at model_uri is packaged with the features, the FeatureStoreClient.score_batch()
# call automatically retrieves the required features from Feature Store before scoring the model.
# The DataFrame returned by score_batch() augments batch_df with
# columns containing the feature values and a column containing model predictions.

fe = FeatureEngineeringClient()

# batch_df has columns ‘customer_id’ and ‘product_id’
predictions = fe.score_batch(
    model_uri=model_uri,
    df=batch_df
)

# The ‘predictions’ DataFrame has these columns:
# ‘customer_id’, ‘product_id’, ‘total_purchases_30d’, ‘category’, ‘prediction’

Úložiště funkcí pracovního prostoru

# Train model
import mlflow
from sklearn import linear_model

feature_lookups = [
    FeatureLookup(
      table_name='recommender_system.customer_features',
      feature_names=['total_purchases_30d'],
      lookup_key='customer_id',
    ),
    FeatureLookup(
      table_name='recommender_system.product_features',
      feature_names=['category'],
      lookup_key='product_id'
    )
  ]

fs = FeatureStoreClient()

with mlflow.start_run():

  # df has columns ['customer_id', 'product_id', 'rating']
  training_set = fs.create_training_set(
    df=df,
    feature_lookups=feature_lookups,
    label='rating',
    exclude_columns=['customer_id', 'product_id']
  )

  training_df = training_set.load_df().toPandas()

  # "training_df" columns ['total_purchases_30d', 'category', 'rating']
  X_train = training_df.drop(['rating'], axis=1)
  y_train = training_df.rating

  model = linear_model.LinearRegression().fit(X_train, y_train)

  fs.log_model(
    model=model,
    artifact_path="recommendation_model",
    flavor=mlflow.sklearn,
    training_set=training_set,
    registered_model_name="recommendation_model"
  )

# Batch inference

# If the model at model_uri is packaged with the features, the FeatureStoreClient.score_batch()
# call automatically retrieves the required features from Feature Store before scoring the model.
# The DataFrame returned by score_batch() augments batch_df with
# columns containing the feature values and a column containing model predictions.

fs = FeatureStoreClient()

# batch_df has columns ‘customer_id’ and ‘product_id’
predictions = fs.score_batch(
    model_uri=model_uri,
    df=batch_df
)

# The ‘predictions’ DataFrame has these columns:
# ‘customer_id’, ‘product_id’, ‘total_purchases_30d’, ‘category’, ‘prediction’

Použití vlastních hodnot funkcí při vyhodnocování modelu zabaleného s metadaty funkcí

Ve výchozím nastavení model zabalený s metadaty funkcí vyhledává funkce z tabulek funkcí při odvozování. Pokud chcete pro bodování použít vlastní hodnoty funkcí, zahrňte je do datového rámce předaného FeatureEngineeringClient.score_batch (pro přípravu funkcí v katalogu Unity) nebo FeatureStoreClient.score_batch (pro úložiště funkcí pracovního prostoru).

Předpokládejme například, že model zabalíte s těmito dvěma funkcemi:

Příprava funkcí v katalogu Unity

feature_lookups = [
    FeatureLookup(
      table_name='ml.recommender_system.customer_features',
      feature_names=['account_creation_date', 'num_lifetime_purchases'],
      lookup_key='customer_id',
    ),
  ]

Úložiště funkcí pracovního prostoru

feature_lookups = [
    FeatureLookup(
      table_name='recommender_system.customer_features',
      feature_names=['account_creation_date', 'num_lifetime_purchases'],
      lookup_key='customer_id',
    ),
  ]

Při odvozování můžete funkci poskytnout vlastní hodnoty account_creation_date voláním score_batch datového rámce, který obsahuje sloupec s názvem account_creation_date. V tomto případě rozhraní API vyhledá pouze num_lifetime_purchases funkci z úložiště funkcí a použije zadané hodnoty vlastních account_creation_date sloupců pro bodování modelu.

Příprava funkcí v katalogu Unity

# batch_df has columns ['customer_id', 'account_creation_date']
predictions = fe.score_batch(
  model_uri='models:/ban_prediction_model/1',
  df=batch_df
)

Úložiště funkcí pracovního prostoru

# batch_df has columns ['customer_id', 'account_creation_date']
predictions = fs.score_batch(
  model_uri='models:/ban_prediction_model/1',
  df=batch_df
)

Trénování a hodnocení modelu pomocí kombinace funkcí Úložiště funkcí a dat umístěných mimo úložiště funkcí

Model můžete vytrénovat pomocí kombinace funkcí úložiště funkcí a dat z externího úložiště funkcí. Když model zabalíte s metadaty funkcí, model načte hodnoty funkcí z úložiště funkcí pro odvozování.

Pokud chcete vytrénovat model, zahrňte do datového rámce předaná FeatureEngineeringClient.create_training_set další data jako sloupce (pro přípravu funkcí v katalogu Unity) nebo FeatureStoreClient.create_training_set (pro úložiště funkcí pracovního prostoru). Tento příklad používá funkci z úložiště funkcí total_purchases_30d a externího sloupce browser.

Příprava funkcí v katalogu Unity

feature_lookups = [
    FeatureLookup(
      table_name='ml.recommender_system.customer_features',
      feature_names=['total_purchases_30d'],
      lookup_key='customer_id',
    ),
  ]

fe = FeatureEngineeringClient()

# df has columns ['customer_id', 'browser', 'rating']
training_set = fe.create_training_set(
  df=df,
  feature_lookups=feature_lookups,
  label='rating',
  exclude_columns=['customer_id']  # 'browser' is not excluded
)

Úložiště funkcí pracovního prostoru

feature_lookups = [
    FeatureLookup(
      table_name='recommender_system.customer_features',
      feature_names=['total_purchases_30d'],
      lookup_key='customer_id',
    ),
  ]

fs = FeatureStoreClient()

# df has columns ['customer_id', 'browser', 'rating']
training_set = fs.create_training_set(
  df=df,
  feature_lookups=feature_lookups,
  label='rating',
  exclude_columns=['customer_id']  # 'browser' is not excluded
)

Při odvozování musí datový rámec použitý ve FeatureStoreClient.score_batch sloupci browser obsahovat.

Příprava funkcí v katalogu Unity

# At inference, 'browser' must be provided
# batch_df has columns ['customer_id', 'browser']
predictions = fe.score_batch(
  model_uri=model_uri,
  df=batch_df
)

Úložiště funkcí pracovního prostoru

# At inference, 'browser' must be provided
# batch_df has columns ['customer_id', 'browser']
predictions = fs.score_batch(
  model_uri=model_uri,
  df=batch_df
)

Načtení modelů a provádění dávkových odvozování pomocí MLflow

Po zaprotokolování modelu pomocí log_model metody FeatureEngineeringClient (pro přípravu funkcí v katalogu Unity) nebo FeatureStoreClient (pro úložiště funkcí pracovního prostoru) lze MLflow použít při odvozování. MLflow.pyfunc.predict načte hodnoty funkcí z úložiště funkcí a také spojí všechny hodnoty zadané v době odvození. Musíte zadat primární klíče funkcí použitých v modelu.

Poznámka:

Dávkové odvození s MLflow vyžaduje MLflow verze 2.11 a vyšší.

# Train model
import mlflow
from sklearn import linear_model

feature_lookups = [
  FeatureLookup(
    table_name='ml.recommender_system.customer_features',
    feature_names=['total_purchases_30d'],
    lookup_key='customer_id',
  ),
  FeatureLookup(
    table_name='ml.recommender_system.product_features',
    feature_names=['category'],
    lookup_key='product_id'
  )
]

fe = FeatureEngineeringClient()

with mlflow.start_run():

  # df has columns ['customer_id', 'product_id', 'rating']
  training_set = fe.create_training_set(
    df=df,
    feature_lookups=feature_lookups,
    label='rating',
    exclude_columns=['customer_id', 'product_id']
  )

  training_df = training_set.load_df().toPandas()

  # "training_df" columns ['total_purchases_30d', 'category', 'rating']
  X_train = training_df.drop(['rating'], axis=1)
  y_train = training_df.rating

  model = linear_model.LinearRegression().fit(X_train, y_train)

  fe.log_model(
    model=model,
    artifact_path="recommendation_model",
    flavor=mlflow.sklearn,
    training_set=training_set,
    registered_model_name="recommendation_model",
    #refers to the default value of "result_type" if not provided at inference
    params={"result_type":"double"},
  )

# Batch inference with MLflow

# NOTE: the result_type parameter can only be used if a default value
# is provided in log_model. This is automatically done for all models
# logged using Databricks Runtime for ML 15.0 or above.
# For earlier Databricks Runtime versions, use set_result as shown below.

# batch_df has columns ‘customer_id’ and ‘product_id’
model = mlflow.pyfunc.load_model(model_version_uri)

# If result_type parameter is provided in log_model
predictions = model.predict(df, {"result_type":"double"})

# If result_type parameter is NOT provided in log_model
model._model_impl.set_result_type("double")
predictions = model.predict(df)