Publish business events using Spark Notebook and react to them using User Data Function (UDF) through Activator

This tutorial guides you through the full workflow of reacting to your own business events in Microsoft Fabric. The steps you follow are:

1. Create a business event and defining its schema
2. Configure a notebook to publish an event using Python.
3. Validate the published events in Real-Time hub.
4. Create an Activator rule that triggers a user data function when the business event occurs.

Create a new business event

1. Go to Business events in Real-Time hub.

2. Select + New business event and then select Create new schema.

   

3. Define the business event schema.

   1. For Name, enter VibrationCriticalDetected.

   2. In the right pane, for Event schema set, select Create.

      

   3. Enter ManufacturingEquipmentHealth for the schema set name.

      

   4. Select Add row in the middle plane.

      

   5. Select string for event type, and enter MachineID for the name.

      

   6. Repeat the above step to add the following properties: ProductionLineID (string), MeasuredVibration (string), ImpactAssessment (string), RecommendationAction (string).

      

   7. Select Next to continue.

4. Review and confirm the configuration, and then select Create to create your business event.

   

Configure notebook for publishing

1. After creating the business event, go to your workspace.

2. Select + New item and then select Notebook in the Analyze and train data section.

   

3. Enter a name for your notebook (for example: BusinessEventTutorialOneNotebook), confirm the location (workspace), and then select Create to create the notebook.

4. Make sure you select either A) Python 3.11 or B) PySpark (Python).

   

   

5. Find the sample code to publish a business event using the following command: notebookutils.businessEvents.help(). You should see output similar to the following one:

   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
   
   

6. Add a new cell, enter the following code, and run it to publish a business event.

   Note

   Make sure to replace the eventSchemaSetWorkspace, eventSchemaSet, and eventTypeName values with the ones you used when creating your business event. The eventData properties should also match the schema you defined for your business event.

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

   Note

   The properties eventSchemaSetWorkspace and eventSchemaSetName support both Fabric item names and Fabric item identifiers (IDs).

   

7. Save your notebook if it's not set to AutoSave.

Verify published business events

1. Go to Business events in Real-Time hub.

2. Select the created business event (for example, VibrationCriticalDetected).

   

3. In the Publisher tab, confirm that you see an event and the notebook is listed in the publisher list.

   

4. Select the Data preview tab.

5. In the publisher filter, select the name of the Notebook previously created publisher.

6. Visualize the event in the preview table.

Configure custom business logic with a user data function

1. Go to your workspace and create a new user data function named ProcessVibrationCritical.

   

2. Create a new Function.

   

3. Modify the logic of your new function and add the input parameters required to receive your business event.

   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()}."
       )
   

4. Test your function.

   1. In the Functions explorer, hover over the function you created, select the ⋯ (three dots) menu, and then select Test.

      

   2. In the Test window, enter these sample values, and select Test.

      • machineID: 12345
      • productionLine: WestLine01
      • measuredVibration: 1.52
      • impactAssessment: Production slowdown risk
      • recommendedAction: Schedule maintenance

   3. Verify the output and the logs to confirm that the function is working as expected.

      

5. Select Publish on the toolbar to publish the function and make it available for use in the Activator rule.

   

Create an Activator trigger to consume events

1. Select the Real-Time hub icon in the left navigation pane of the Fabric portal.

2. In the Real-Time hub, select Business events under the Subscribe to category.

3. In the Business events list, locate VibrationCriticalDetected event. Select either the ⚡ lightning icon or the ⋯ (three dots) menu next to the event, and then select Set alert.

   

4. On the Add rule page, in the Details section, for Rule name, enter a name for the rule. For example, VibrationCriticalDetected_Rule.

5. In the Condition section, for Check, select On each event.

6. In the Action section, select one of the following actions. To configure the alert to trigger a function when the condition is met, follow these steps:

   1. For Select action, select Run Function.

      

   2. Select the Fabric item you want to run, and then select Add to continue (for example, the ProcessVibrationCritical user data function).

      

   3. Select the function you want to use to process the action (for example, the processVibrationCritical function).

      

   4. Map each input parameter defined in the function with the business event property you previously defined. Type @ and select the business event property from the dropdown list to map it to the function parameter.

      

      Repeat this process for each input parameter defined in the function.

      

7. In the Save location section, for Workspace, select the workspace where you want to create the Fabric activator item.

8. For Item, select the drop-down list, and then select Create a new item.

   

9. In the dialog box, enter a name for the new Fabric activator item (for example, VibrationCriticalDetected_Activator), and then select Create.

   

10. You see the Alert created page with a link to open the rule in the Fabric activator user interface in a separate tab.

    

11. Select Open to open the rule in the Fabric Activator user interface.

Test the solution

Publish another event

In the Spark notebook, run the cell to publish a new VibrationCriticalDetected business event again.

View business event in the Real-Time hub

Publishers tab

1. In the Real-Time hub, select Business events in the left menu.

2. Select VibrationCriticalDetected from the list of business events.

3. In the Publisher tab, confirm that you see the newly published event and the old event you published at the beginning of the tutorial. If you don't see the new event, select the refresh button to refresh the list of events. It might take a few seconds for the new event to appear in the list.

4. In the list of publishers, confirm that the notebook you used to publish the event is listed as a publisher.

   

Consumers tab

1. Switch to the Consumers tab.

2. Confirm that you see an event was delivered within the last hour.

3. Confirm that the activator you created is listed as a consumer.

   

Data preview tab

1. Switch to the Data preview tab.

2. Confirm that you see the consumer filter with the activator you created.

   

View business event in the activator run history

1. Go to the workspace where you created the activator item, and open the activator item (for example, VibrationCriticalDetected_Activator) if it's not already open.

2. In the activator item, select the History tab.

3. Confirm that you see one activation.

   

View logs for the user data function

1. In the workspace, open the ProcessVibrationCritical user data function you created.

2. Switch to Run only mode using the mode switcher.

   

3. Hover over the function name in the functions list, select the ellipses icon (...), and then select View historical log.

   

4. You see table with the historical runs of the function. Select the most recent run to view the logs for that run.

   

5. Confirm that the logs show the event payload in the JSON format as you defined in the function logic, which indicates that the function was triggered by the business event and processed the event data correctly.