Delen via

Functies gebruiken om modellen te trainen

  • Artikel

In dit artikel wordt beschreven hoe u modellen kunt trainen met behulp van Functie-engineering in Unity Catalog of de lokale werkruimtefunctieopslag. U moet eerst een trainingsgegevensset maken, waarmee de functies worden gedefinieerd die moeten worden gebruikt en hoe u deze kunt koppelen. Wanneer u vervolgens een model traint, behoudt het model verwijzingen naar de functies.

Wanneer u een model traint met functie-engineering in Unity Catalog, kunt u de herkomst van het model bekijken in Catalog Explorer. Tabellen en functies die zijn gebruikt om het model te maken, worden automatisch bijgehouden en weergegeven. Zie Herkomst van functiearchief weergeven.

Wanneer u het model gebruikt voor deductie, kunt u ervoor kiezen om functiewaarden op te halen uit het functiearchief. U kunt het model ook bedienen met Model Serving en de functies worden automatisch opzoekfuncties gepubliceerd in online winkels. Onderdelenarchiefmodellen zijn ook compatibel met de pyfunc-interface van MLflow, zodat u MLflow kunt gebruiken om batchdeductie uit te voeren met functietabellen.

Als uw model omgevingsvariabelen gebruikt, leest u hoe u deze kunt gebruiken wanneer u het model online bedient op Toegang tot resources configureren vanuit eindpunten voor het leveren van modellen.

Een model kan maximaal 50 tabellen en 100 functies gebruiken voor training.

Een trainingsgegevensset selecteren

Als u specifieke functies wilt selecteren in een functietabel voor modeltraining, maakt u een trainingsgegevensset met behulp van de API (voor Functie-engineering in Unity Catalog) of FeatureStoreClient.create_training_set (voor werkruimtefunctiearchief) en een object met de FeatureEngineeringClient.create_training_set naam aFeatureLookup. Een FeatureLookup geeft aan welke functie in de trainingsset moet worden gebruikt, inclusief de naam van de functietabel, de naam(en) van de functies en de sleutel(en) die moeten worden gebruikt bij het samenvoegen van de functietabel met het DataFrame dat is doorgegeven aan create_training_set. Zie Functiezoekactie voor meer informatie.

Gebruik de feature_names parameter wanneer u een FeatureLookup. feature_names neemt één functienaam, een lijst met functienamen of Geen om alle functies (met uitzondering van primaire sleutels) in de functietabel op te zoeken op het moment dat de trainingsset wordt gemaakt.

Notitie

Het type en de volgorde van lookup_key kolommen in het DataFrame moeten overeenkomen met het type en de volgorde van de primaire sleutels (met uitzondering van tijdstempelsleutels) van de referentiefunctietabel.

Dit artikel bevat codevoorbeelden voor beide versies van de syntaxis.

In dit voorbeeld bevat het DataFrame dat wordt trainingSet.load_df geretourneerd door een kolom voor elke functie in feature_lookups. Alle kolommen van het DataFrame worden bewaard, create_training_set behalve de kolommen die zijn uitgesloten met behulp van exclude_columns.

Functie-engineering in Unity Catalog

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()

Werkruimtefunctiearchief

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()

Een TrainingSet maken wanneer opzoeksleutels niet overeenkomen met de primaire sleutels

Gebruik het argument lookup_key in de FeatureLookup kolomnaam in de trainingsset. create_training_set voert een geordende join uit tussen de kolommen uit de trainingsset die is opgegeven in het lookup_key argument met behulp van de volgorde waarin de primaire sleutels zijn opgegeven toen de functietabel werd gemaakt.

In dit voorbeeld recommender_system.customer_features heeft u de volgende primaire sleutels: customer_id, dt.

De recommender_system.product_features functietabel heeft de primaire sleutel product_id.

Als de training_df volgende kolommen hebben:

  • cid
  • transaction_dt
  • product_id
  • rating

met de volgende code worden de juiste functiezoekacties voor het TrainingSetvolgende gemaakt:

Functie-engineering in Unity Catalog

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'
    )
  ]

Werkruimtefunctiearchief

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'
    )
  ]

Wanneer create_training_set deze wordt aangeroepen, wordt er een trainingsgegevensset gemaakt door een left join uit te voeren, de tabellen recommender_system.customer_features samen te voegen en training_df de sleutels (customer_id,dt) te gebruiken die overeenkomen met (cid,transaction_dt), zoals wordt weergegeven in de volgende code:

Functie-engineering in Unity Catalog

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"
)

Werkruimtefunctiearchief

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"
)

Een TrainingSet maken met twee functies met dezelfde naam uit verschillende functietabellen

Gebruik het optionele argument output_name in de FeatureLookup. De opgegeven naam wordt gebruikt in plaats van de functienaam in het DataFrame dat wordt geretourneerd door TrainingSet.load_df. Met de volgende code bevat bijvoorbeeld het DataFrame dat wordt geretourneerd door training_set.load_df kolommen customer_height en product_height.

Functie-engineering in Unity Catalog

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()

Werkruimtefunctiearchief

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()

Een TrainingSet maken met dezelfde functie meerdere keren

Als u een TrainingSet wilt maken met dezelfde functie die is gekoppeld aan verschillende opzoeksleutels, gebruikt u meerdere FeatureLookups. Gebruik een unieke output_name functie voor elke FeatureLookup-uitvoer.

Functie-engineering in Unity Catalog

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'
    )
  ]

Werkruimtefunctiearchief

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'
    )
  ]

Een TrainingSet maken voor machine learning-modellen zonder supervisie

Instellen label=None bij het maken van een TrainingSet voor leermodellen zonder supervisie. De volgende TrainingSet kan bijvoorbeeld worden gebruikt om verschillende klanten in groepen te clusteren op basis van hun interesses:

Functie-engineering in Unity Catalog

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()

Werkruimtefunctiearchief

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()

Modellen trainen en batchdeductie uitvoeren met functietabellen

Wanneer u een model traint met behulp van functies uit Feature Store, behoudt het model verwijzingen naar de functies. Wanneer u het model gebruikt voor deductie, kunt u ervoor kiezen om functiewaarden op te halen uit de Functieopslag. U moet de primaire sleutel(s) opgeven van de functies die in het model worden gebruikt. Het model haalt de functies op die het nodig heeft uit de Feature Store in uw werkruimte. Vervolgens worden de functiewaarden toegevoegd als dat nodig is tijdens het scoren.

Functie opzoeken op deductietijd ondersteunen:

  • U moet het model registreren met behulp van FeatureEngineeringClient de log_model methode (voor Functie-engineering in Unity Catalog) of FeatureStoreClient (voor werkruimtefunctiearchief).
  • U moet het DataFrame gebruiken dat wordt geretourneerd door TrainingSet.load_df het model te trainen. Als u dit DataFrame op enigerlei wijze wijzigt voordat u het model gebruikt om het model te trainen, worden de wijzigingen niet toegepast wanneer u het model gebruikt voor deductie. Dit vermindert de prestaties van het model.
  • Het modeltype moet een overeenkomend python_flavor model hebben in MLflow. MLflow ondersteunt de meeste Python-modeltrainingsframeworks, waaronder:
    • scikit-learn
    • keras
    • PyTorch
    • SparkML
    • LightGBM
    • XGBoost
    • TensorFlow Keras (met behulp van de python_flavor mlflow.keras)
  • Aangepaste MLflow pyfunc-modellen

Functie-engineering in Unity Catalog

# 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’

Werkruimtefunctiearchief

# 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’

Aangepaste functiewaarden gebruiken bij het scoren van een model dat is verpakt met metagegevens van functies

Standaard zoekt een model dat is verpakt met functiemetagegevens functies uit functietabellen bij deductie. Als u aangepaste functiewaarden wilt gebruiken voor scoren, neemt u deze op in het DataFrame dat is doorgegeven aan FeatureEngineeringClient.score_batch (voor Functie-engineering in Unity Catalog) of FeatureStoreClient.score_batch (voor werkruimtefunctiearchief).

Stel dat u een model met deze twee functies inpakt:

Functie-engineering in Unity Catalog

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

Werkruimtefunctiearchief

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

Bij deductie kunt u aangepaste waarden voor de functie account_creation_date opgeven door een DataFrame aan te roepen score_batch dat een kolom met de naam account_creation_datebevat. In dit geval zoekt de API alleen de num_lifetime_purchases functie uit de Feature Store op en gebruikt de opgegeven aangepaste account_creation_date kolomwaarden voor het scoren van modellen.

Functie-engineering in Unity Catalog

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

Werkruimtefunctiearchief

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

Een model trainen en beoordelen met behulp van een combinatie van functies en gegevens die zich buiten de functieopslag bevinden

U kunt een model trainen met behulp van een combinatie van functies en gegevens van buiten het functiearchief. Wanneer u het model inpakt met metagegevens van functies, haalt het model functiewaarden op uit het functiearchief voor deductie.

Als u een model wilt trainen, moet u de extra gegevens opnemen als kolommen in het DataFrame dat wordt doorgegeven aan FeatureEngineeringClient.create_training_set (voor Functie-engineering in Unity Catalog) of FeatureStoreClient.create_training_set (voor werkruimtefunctiearchief). In dit voorbeeld wordt de functie total_purchases_30d uit Feature Store en de externe kolom gebruikt browser.

Functie-engineering in Unity Catalog

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
)

Werkruimtefunctiearchief

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
)

Bij deductie moet het DataFrame dat wordt FeatureStoreClient.score_batch gebruikt, de browser kolom bevatten.

Functie-engineering in Unity Catalog

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

Werkruimtefunctiearchief

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

Modellen laden en batchdeductie uitvoeren met behulp van MLflow

Nadat een model is geregistreerd met behulp van de log_model methode (voor Functie-engineering in Unity Catalog) of FeatureStoreClient (voor werkruimtefunctiearchief), kan MLflow worden gebruikt bij deductie FeatureEngineeringClient . MLflow.pyfunc.predict haalt functiewaarden op uit het Functiearchief en voegt ook alle waarden toe die tijdens deductietijd worden opgegeven. U moet de primaire sleutel(s) opgeven van de functies die in het model worden gebruikt.

Notitie

Batchdeductie met MLflow vereist MLflow versie 2.11 en hoger. Modellen die gebruikmaken van tijdreeksfunctietabellen worden niet ondersteund. Als u batchdeductie wilt uitvoeren met tijdreeksfunctietabellen, gebruikt u score_batch. Zie Modellen trainen en batchdeductie uitvoeren met functietabellen.

# 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)