Руководство 5. Разработка набора компонентов с помощью пользовательского источника

Статья
11/21/2024

Машинное обучение Azure хранилище управляемых функций позволяет обнаруживать, создавать и использовать функции. Функции служат соединительной тканью в жизненном цикле машинного обучения, начиная с этапа прототипирования, где вы экспериментируйте с различными функциями. Этот жизненный цикл продолжается на этапе эксплуатации, где вы развертываете модели и выполняете поиск данных функций. Дополнительные сведения о хранилищах компонентов см. в ресурсе концепции хранилища компонентов.

В части 1 этого руководства показано, как создать спецификацию набора компонентов с пользовательскими преобразованиями, включить материализацию и выполнить обратную заполнение. Часть 2 показала, как экспериментировать с функциями в потоках экспериментирования и обучения. Часть 3 объяснила повторяющуюся материализацию для transactions набора компонентов и показала, как запустить конвейер вывода пакетной службы в зарегистрированной модели. Часть 4 описывает выполнение пакетного вывода.

В этом руководстве вы узнаете, как это сделать

Определите логику для загрузки данных из пользовательского источника данных.
Настройте и зарегистрируйте набор компонентов для использования из этого пользовательского источника данных.
Проверьте зарегистрированный набор компонентов.

Предварительные требования

Примечание.

В этом руководстве используется записная книжка Машинное обучение Azure с бессерверными вычислениями Spark.

Обязательно выполните предыдущие учебники в этой серии. В этом руководстве повторно используется хранилище функций и другие ресурсы, созданные в этих предыдущих руководствах.

Настройка

В этом руководстве используется основной пакет SDK для хранилища компонентов Python (azureml-featurestore). Пакет SDK для Python используется для операций создания, чтения, обновления и удаления (CRUD) в хранилищах компонентов, наборах компонентов и сущностях хранилища компонентов.

Вам не нужно явно устанавливать эти ресурсы для этого руководства, conda.yml так как в приведенных здесь инструкциях по настройке файл охватывает их.

Настройка записной книжки Машинное обучение Azure Spark

Вы можете создать записную книжку и выполнить инструкции, описанные в этом руководстве, пошаговые инструкции. Вы также можете открыть и запустить существующую записную книжку featurestore_sample/notebooks/sdk_only/5.Develop-feature-set-custom-source.ipynb. Оставьте это руководство открытым и ознакомьтесь с ним по ссылкам документации и дополнительным пояснениям.

В верхнем меню в раскрывающемся списке вычислений выберите бессерверные вычисления Spark в разделе Машинное обучение Azure Бессерверный Spark.
Настройка сеанса:
1. Выберите " Настройка сеанса " в верхней строке состояния
2. Выберите вкладку "Пакеты Python", выберите "Отправить файл Conda"
3. Выбор файла Upload Conda
4. Отправка файла conda.yml, отправленного в первом руководстве
5. При необходимости увеличьте время ожидания сеанса (время простоя), чтобы избежать частого повторного выполнения предварительных требований.

Настройка корневого каталога для примеров

Эта ячейка кода настраивает корневой каталог для примеров. Для установки всех зависимостей и запуска сеанса Spark требуется около 10 минут.

import os

# Please update the dir to ./Users/{your_user_alias} (or any custom directory you uploaded the samples to).
# You can find the name from the directory structure in the left navigation panel.
root_dir = "./Users/<your_user_alias>/featurestore_sample"

if os.path.isdir(root_dir):
    print("The folder exists.")
else:
    print("The folder does not exist. Please create or fix the path")

Инициализация клиента CRUD рабочей области хранилища компонентов

Инициализировать MLClient рабочую область хранилища компонентов, чтобы охватить операции создания, чтения, обновления и удаления (CRUD) в рабочей области хранилища компонентов.

from azure.ai.ml import MLClient
from azure.ai.ml.identity import AzureMLOnBehalfOfCredential

# Feature store
featurestore_name = (
    "<FEATURESTORE_NAME>"  # use the same name that was used in the tutorial #1
)
featurestore_subscription_id = os.environ["AZUREML_ARM_SUBSCRIPTION"]
featurestore_resource_group_name = os.environ["AZUREML_ARM_RESOURCEGROUP"]

# Feature store ml client
fs_client = MLClient(
    AzureMLOnBehalfOfCredential(),
    featurestore_subscription_id,
    featurestore_resource_group_name,
    featurestore_name,
)

Инициализация основного клиента пакета SDK для хранилища компонентов

Как упоминалось ранее, в этом руководстве используется основной пакет SDK для хранилища компонентов Python (azureml-featurestore). Этот инициализированный клиент ПАКЕТА SDK охватывает операции создания, чтения, обновления и удаления (CRUD) в хранилищах компонентов, наборах компонентов и сущностях хранилища компонентов.

from azureml.featurestore import FeatureStoreClient
from azure.ai.ml.identity import AzureMLOnBehalfOfCredential

featurestore = FeatureStoreClient(
    credential=AzureMLOnBehalfOfCredential(),
    subscription_id=featurestore_subscription_id,
    resource_group_name=featurestore_resource_group_name,
    name=featurestore_name,
)

Пользовательское определение источника

Вы можете определить собственную логику загрузки источника из любого хранилища данных, имеющего пользовательское определение источника. Реализуйте класс определяемой пользователем функции (UDF) исходного процессора (CustomSourceTransformer в этом руководстве) для использования этой функции. Этот класс должен определять функцию __init__(self, **kwargs) и process(self, start_time, end_time, **kwargs) функцию. Словарь kwargs предоставляется как часть определения спецификации набора компонентов. Затем это определение передается в UDF. Вычисляются start_time и end_time передаются в функцию UDF.

Это пример кода для класса UDF исходного процессора:

from datetime import datetime

class CustomSourceTransformer:
    def __init__(self, **kwargs):
        self.path = kwargs.get("source_path")
        self.timestamp_column_name = kwargs.get("timestamp_column_name")
        if not self.path:
            raise Exception("`source_path` is not provided")
        if not self.timestamp_column_name:
            raise Exception("`timestamp_column_name` is not provided")

    def process(
        self, start_time: datetime, end_time: datetime, **kwargs
    ) -> "pyspark.sql.DataFrame":
        from pyspark.sql import SparkSession
        from pyspark.sql.functions import col, lit, to_timestamp

        spark = SparkSession.builder.getOrCreate()
        df = spark.read.json(self.path)

        if start_time:
            df = df.filter(col(self.timestamp_column_name) >= to_timestamp(lit(start_time)))

        if end_time:
            df = df.filter(col(self.timestamp_column_name) < to_timestamp(lit(end_time)))

        return df

Создание спецификации набора компонентов с пользовательским источником и локальное экспериментирование с ним

Теперь создайте спецификацию набора компонентов с пользовательским определением источника и используйте ее в среде разработки для экспериментов с набором компонентов. Записная книжка учебника, подключенная к бессерверным вычислениям Spark, служит средой разработки.

from azureml.featurestore import create_feature_set_spec
from azureml.featurestore.feature_source import CustomFeatureSource
from azureml.featurestore.contracts import (
    SourceProcessCode,
    TransformationCode,
    Column,
    ColumnType,
    DateTimeOffset,
    TimestampColumn,
)

transactions_source_process_code_path = (
    root_dir
    + "/featurestore/featuresets/transactions_custom_source/source_process_code"
)
transactions_feature_transform_code_path = (
    root_dir
    + "/featurestore/featuresets/transactions_custom_source/feature_process_code"
)

udf_featureset_spec = create_feature_set_spec(
    source=CustomFeatureSource(
        kwargs={
            "source_path": "wasbs://data@azuremlexampledata.blob.core.windows.net/feature-store-prp/datasources/transactions-source-json/*.json",
            "timestamp_column_name": "timestamp",
        },
        timestamp_column=TimestampColumn(name="timestamp"),
        source_delay=DateTimeOffset(days=0, hours=0, minutes=20),
        source_process_code=SourceProcessCode(
            path=transactions_source_process_code_path,
            process_class="source_process.CustomSourceTransformer",
        ),
    ),
    feature_transformation=TransformationCode(
        path=transactions_feature_transform_code_path,
        transformer_class="transaction_transform.TransactionFeatureTransformer",
    ),
    index_columns=[Column(name="accountID", type=ColumnType.string)],
    source_lookback=DateTimeOffset(days=7, hours=0, minutes=0),
    temporal_join_lookback=DateTimeOffset(days=1, hours=0, minutes=0),
    infer_schema=True,
)

udf_featureset_spec

Затем определите окно компонента и отобразите значения признаков в этом окне.

from datetime import datetime

st = datetime(2023, 1, 1)
et = datetime(2023, 6, 1)

display(
    udf_featureset_spec.to_spark_dataframe(
        feature_window_start_date_time=st, feature_window_end_date_time=et
    )
)

Экспорт в виде спецификации набора компонентов

Чтобы зарегистрировать спецификацию набора компонентов в хранилище компонентов, сначала сохраните эту спецификацию в определенном формате. Просмотрите спецификацию созданного transactions_custom_source набора компонентов. Откройте этот файл из дерева файлов, чтобы просмотреть спецификацию: featurestore/featuresets/transactions_custom_source/spec/FeaturesetSpec.yaml

Спецификация содержит следующие элементы:

features: список функций и их типов данных.
index_columns: ключи соединения, необходимые для доступа к значениям из набора компонентов.

Дополнительные сведения о спецификации см. в статье "Основные сведения о сущностях верхнего уровня" в хранилище управляемых функций и cli (версии 2) с набором ресурсов схемы YAML.

Сохраняемость спецификации набора компонентов обеспечивает еще одно преимущество: спецификация набора компонентов может управляться источником.

feature_spec_folder = (
    root_dir + "/featurestore/featuresets/transactions_custom_source/spec"
)

udf_featureset_spec.dump(feature_spec_folder)

Регистрация набора функций транзакции в хранилище компонентов

Этот код используется для регистрации ресурса набора компонентов, загруженного из пользовательского источника в хранилище компонентов. Затем вы можете повторно использовать этот ресурс и легко поделиться им. Регистрация ресурса набора компонентов предоставляет управляемые возможности, включая управление версиями и материализацию.

from azure.ai.ml.entities import FeatureSet, FeatureSetSpecification

transaction_fset_config = FeatureSet(
    name="transactions_custom_source",
    version="1",
    description="transactions feature set loaded from custom source",
    entities=["azureml:account:1"],
    stage="Development",
    specification=FeatureSetSpecification(path=feature_spec_folder),
    tags={"data_type": "nonPII"},
)

poller = fs_client.feature_sets.begin_create_or_update(transaction_fset_config)
print(poller.result())

Получите зарегистрированный набор компонентов и распечатайте связанные сведения.

# Look up the feature set by providing name and version
transactions_fset_config = featurestore.feature_sets.get(
    name="transactions_custom_source", version="1"
)
# Print feature set information
print(transactions_fset_config)

Тестирование создания компонентов из зарегистрированного набора компонентов

to_spark_dataframe() Используйте функцию набора компонентов, чтобы протестировать поколение компонентов из зарегистрированного набора компонентов и отобразить компоненты. print-txn-fset-sample-values

df = transactions_fset_config.to_spark_dataframe()
display(df)

Вы сможете успешно получить зарегистрированный набор компонентов в виде кадра данных Spark, а затем отобразить его. Теперь эти функции можно использовать для соединения между точками во времени с данными наблюдения и последующими шагами в конвейере машинного обучения.

Очистка

Если вы создали группу ресурсов для учебника, вы можете удалить эту группу ресурсов, которая удаляет все ресурсы, связанные с этим руководством. В противном случае можно удалить ресурсы по отдельности:

Чтобы удалить хранилище компонентов, откройте группу ресурсов в портал Azure, выберите хранилище компонентов и удалите его.
Назначаемое пользователем управляемое удостоверение (UAI), назначенное рабочей области хранилища компонентов, не удаляется при удалении хранилища компонентов. Чтобы удалить UAI, следуйте этим инструкциям.
Чтобы удалить автономное хранилище типа учетной записи хранения, откройте группу ресурсов в портал Azure, выберите созданное хранилище и удалите его.
Чтобы удалить экземпляр Кэш Azure для Redis, откройте группу ресурсов в портал Azure, выберите созданный экземпляр и удалите его.

Поделиться через