Публикация бизнес-событий с помощью записной книжки Spark и реагирование на них с помощью функции пользовательских данных (UDF) с помощью активатора

В этом руководстве описан полный рабочий процесс реагирования на собственные бизнес-события в Microsoft Fabric. Ниже приведены следующие действия.

Создание бизнес-события и определение схемы
Настройте записную книжку для публикации события с помощью Python.
Проверьте опубликованные события в центре Real-Time.
Создайте правило активатора, которое активирует функцию пользовательских данных при возникновении бизнес-события.

Это важно

Эта функция доступна в предварительной версии.

Создание нового бизнес-события

Перейдите к бизнес-событиям в центре Real-Time.
Выберите +Создать бизнес-событие и нажмите кнопку "Создать схему".
Определите схему бизнес-событий.
1. Для параметра Имя введите VibrationCriticalDetected.
2. В правой области для набора схем событий нажмите кнопку "Создать".
3. Введите ManufacturingEquipmentHealth для имени набора схем.
4. Выберите "Добавить строку " в средней плоскости.
5. Выберите строку для типа события и введите MachineIDимя.
6. Повторите приведенный выше шаг, чтобы добавить следующие свойства: ProductionLineID (строка), MeasuredVibration (строка), ImpactAssessment (строка), RecommendationAction (строка).
7. Для продолжения выберите Далее.
Проверьте и подтвердите конфигурацию, а затем нажмите кнопку "Создать ", чтобы создать бизнес-событие.

Настройка записной книжки для публикации

После создания бизнес-мероприятия перейдите в свою рабочую область.
Выберите +Создать элемент и выберите "Записная книжка " в разделе "Анализ и обучение данных ".
Введите имя записной книжки (например, BusinessEventTutorialOneNotebook), подтвердите расположение (рабочая область), а затем нажмите кнопку "Создать ", чтобы создать записную книжку.
Убедитесь, что вы выбрали Python 3.11 или B) PySpark (Python).

Найдите пример кода для публикации бизнес-события с помощью следующей команды. notebookutils.businessEvents.help() Вы должны увидеть выходные данные, аналогичные следующему:

Help on module notebookutils.businessEvents in notebookutils:

NAME
    notebookutils.businessEvents - [Preview] Utility for Business Events operations in Fabric

FUNCTIONS
    help(methodName: str = '') -> None
        [Preview] Provides help for the notebookutils.businessEvents module or the specified method.

        Examples:
        notebookutils.businessEvents.help()
        notebookutils.businessEvents.help("publish")
        :param methodName: The name of the method to get help with.

    publish(eventSchemaSetWorkspace: str, eventSchemaSet: str, eventTypeName: str, eventData: Union[Dict[str, Any], List[Dict[str, Any]]], dataVersion: str = 'v1') -> bool
        [Preview] Publish business events data to the specified event type.

        Examples:
        notebookutils.businessEvents.publish(
            eventSchemaSetWorkspace="my-workspace-id",
            eventSchemaSet="OrderEvents",
            eventTypeName="OrderDelayed",
            eventData={"orderId": "12345", "status": "delayed", "reason": "weather"},
            dataVersion="v1"
        )

        # Batch publish multiple events
        notebookutils.businessEvents.publish(
            eventSchemaSetWorkspace="my-workspace-id",
            eventSchemaSet="OrderEvents",
            eventTypeName="OrderDelayed",
            eventData=[
                {"orderId": "12345", "status": "delayed", "reason": "weather"},
                {"orderId": "12346", "status": "delayed", "reason": "traffic"}
            ],
            dataVersion="v1"
        )

        :param eventSchemaSetWorkspace: The workspace ID or name where the event schema set is located
        :param eventSchemaSet: The ID or name of the event schema set
        :param eventTypeName: The name of the business events type to publish to
        :param eventData: The event data payload as a dictionary or list of dictionaries for batch publishing
        :param dataVersion: The version of the event type schema (default: "v1")
        :return: True if the event was published successfully
        :raises: Exception if the event could not be published

DATA
    __all__ = ['help', 'publish']

FILE
    /home/trusted-service-user/jupyter-env/python3.11/lib/python3.11/site-packages/notebookutils/businessEvents.py

Добавьте новую ячейку, введите следующий код и запустите его для публикации бизнес-события.

Замечание

Обязательно замените значения eventSchemaSetWorkspace, eventSchemaSet, и eventTypeName на те, которые вы использовали при создании вашего бизнес-события. Свойства eventData также должны соответствовать схеме, определенной для вашего бизнес-события.
```
notebookutils.businessEvents.publish(
    eventSchemaSetWorkspace="My workspace",
    eventSchemaSet="ManufacturingEquipmentHealth",
    eventTypeName="VibrationCriticalDetected",
    eventData={
        "MachineID": "12345",
        "ProductionLineID": "WestLine01",
        "MeasuredVibration": "1.52",
        "ImpactAssessment": "Production slowdown risk",
        "RecommendationAction": "Schedule maintenance"
    },
    dataVersion="v1"
)
```
Замечание

Свойства eventSchemaSetWorkspace и eventSchemaSetName поддерживают как имена элементов Fabric, так и идентификаторы (ID) элементов Fabric.
Сохраните записную книжку, если не задано значение AutoSave.

Проверка опубликованных бизнес-событий

Перейдите к бизнес-событиям в центре Real-Time.
Выберите созданное бизнес-событие (например, VibrationCriticalDetected).
На вкладке "Издатель" подтвердите, что вы видите событие и записная книжка указана в списке издателя.
Перейдите на вкладку "Предварительный просмотр данных ".
В фильтре издателя выберите имя предварительно созданного издателя Ноутбук.
Визуализировать событие в таблице предварительного просмотра.

Настройка пользовательской бизнес-логики с помощью функции пользовательских данных

Перейдите в рабочую область и создайте новую функцию данных пользователя с именем ProcessVibrationCritical.
Создайте новую функцию.

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

import datetime
import fabric.functions as fn
import logging
import json

udf = fn.UserDataFunctions()

@udf.function()

def processVibrationCritical(
    machineID: str,
    productionLine: str,
    measuredVibration: str,
    impactAssessment: str,
    recommendedAction: str
) -> str:

    logging.info("processVibrationCritical invoked.")

    event_data = {
        "machineID": machineID,
        "productionLine": productionLine,
        "measuredVibration": measuredVibration,
        "impactAssessment": impactAssessment,
        "recommendedAction": recommendedAction
    }

    # Log as structured JSON for easy searching/filtering in logs
    logging.info("processVibrationCritical payload=%s", json.dumps(event_data))

    return (
        f"Processed processVibrationCritical for machineID={machineID} "
        f"on line={productionLine} at {datetime.datetime.now()}."
    )

Протестируйте функцию.
1. В обозревателе функций наведите указатель мыши на созданную функцию, выберите меню ⋯(три точки), а затем выберите "Тест".
2. В окне "Тест" введите эти примеры значений и выберите "Тест".
  - machineID: 12345
  - линияПроизводства: WestLine01
  - измереннаяВибрация: 1.52
  - Оценка воздействия: риск замедления производства
  - recommendedAction: планирование обслуживания
3. Проверьте выходные данные и журналы, чтобы убедиться, что функция работает должным образом.
Выберите "Опубликовать " на панели инструментов, чтобы опубликовать функцию и сделать ее доступной для использования в правиле активации.

Создайте триггер активатора для использования событий

Выберите значок центра в реальном времени в области навигации слева на портале Fabric.
В центре Real-Time выберите "Бизнес-события " в категории "Подписка на ".
В списке бизнес-событий найдите VibrationCriticalDetected событие. Выберите значок молнии или меню ⋯(три точки) рядом ⚡ с событием, а затем нажмите кнопку "Задать оповещение".
На странице "Добавление правила " в разделе "Сведения " введите имя правила. Например, VibrationCriticalDetected_Rule.
В разделе «Условие» для Проверка выберите «На каждом событии».
В разделе "Действие " выберите одно из следующих действий. Чтобы настроить оповещение для активации функции при выполнении условия, выполните следующие действия.
1. Для Выбор действия, выберите Выполнить функцию.
2. Выберите элемент Fabric, который вы хотите запустить, и нажмите кнопку "Добавить для продолжения" (например, функция пользовательских ProcessVibrationCritical данных).
3. Выберите функцию, которую нужно использовать для обработки действия (например, processVibrationCritical функции).
4. Сопоставляйте каждый входной параметр, определенный в функции, с помощью свойства бизнес-события, которое вы ранее определили. Введите @ и выберите свойство бизнес-события из раскрывающегося списка, чтобы сопоставить его с параметром функции.
  
  Повторите этот процесс для каждого входного параметра, определенного в функции.
В разделе "Сохранить расположение " для рабочей области выберите рабочую область, в которой нужно создать элемент активатора Fabric.
Для элемента выберите раскрывающийся список и нажмите кнопку "Создать элемент".
В диалоговом окне введите имя нового элемента активатора Fabric (например, VibrationCriticalDetected_Activator), а затем нажмите кнопку "Создать".
Вы увидите страницу создания оповещения со ссылкой, чтобы открыть правило в пользовательском интерфейсе активатора Fabric на отдельной вкладке.
Выберите Открыть, чтобы открыть правило в пользовательском интерфейсе Fabric Activator.

Тестирование решения

Публикация другого события

В записной книжке Spark снова запустите ячейку, чтобы опубликовать новое VibrationCriticalDetected бизнес-событие.

Просмотр бизнес-события в центре Real-Time

Вкладка "Издатели"

В центре Real-Time выберите бизнес-события в меню слева.
Выберите VibrationCriticalDetected из списка бизнес-событий.
На вкладке "Издатель" убедитесь, что вы видите недавно опубликованное событие и старое событие, опубликованное в начале руководства. Если новое событие не отображается, нажмите кнопку обновления, чтобы обновить список событий. Чтобы новое событие отображалось в списке, может потребоваться несколько секунд.
В списке издателей убедитесь, что записная книжка, используемая для публикации события, указана в качестве издателя.

Вкладка "Потребители"

Перейдите на вкладку "Потребители".
Проверьте, что вы видите, что событие было доставлено в течение последнего часа.
Убедитесь, что созданный активатор указан как потребитель.

Вкладка предварительного просмотра данных

Перейдите на вкладку предварительного просмотра данных .
Убедитесь, что вы видите потребительский фильтр с активатором, который вы создали.

Просмотр бизнес-события в журнале выполнения активатора

Перейдите в рабочую область, в которой вы создали элемент активатора, и откройте элемент активатора (например, VibrationCriticalDetected_Activator), если он еще не открыт.
В элементе активатора выберите вкладку «Журнал».
Убедитесь, что вы видите одну активацию.

Просмотр журналов для функции данных пользователя

В рабочей области откройте созданную вами функцию ProcessVibrationCritical данных пользователя.
Переключитесь на режим запуска только с помощью переключателя режима.
Наведите указатель мыши на имя функции в списке функций, выберите значок многоточия (...), а затем выберите "Просмотреть исторический журнал".
Вы увидите таблицу с историческими запусками функции. Выберите последний запуск, чтобы просмотреть журналы для этого запуска.
Убедитесь, что в журналах отображаются полезные данные события в формате JSON, как определено в логике функции, что означает, что функция была активирована бизнес-событием и правильно обрабатывала данные события.

Обратная связь

Были ли сведения на этой странице полезными?

Last updated on 2026-03-18