Lab 4: Automate data ingestion and alerting with Activator
Learn how to build the Silver layer of your Medallion architecture and keep it updated automatically using Update Policies, then configure real‑time alerts with Activator to detect operational issues in bike stations as they happen
This is the lab 4 out of 5 of the series
Hector Sven
- ⏮️Previous Lab 3: Enrich your real-time data stream with external data
- ⏭️Next Lab 5: Real-Time Dashboards
Contents
- Introduction
- Medallion Framework within KQL Databases
- Task 1: Create Silver Tables
- Task 2: Initial load of silver table
- Task 3: Create Update Policies to Load the Silver Layer
- Alerts with Activator
- Task 4: Creating Alerts with Activator
Introduction
In this lab, you will take the raw data (Bronze layer) created in the previous lab and build out the Silver layer of your Medallion Framework and keep the data up-to-date via Policies. You will then set up an alerting system using Activator
Medallion Framework within KQL Databases
Task 1: Create Silver Tables
You will be creating a data enriched table in your silver layer. Below you can see each table and columns that are added during the enrichment process.
#1. Navigate to the KQL Database you have been using within your Fabric workspace 1️⃣, from the menu bar at the top select the KQL Queryset 2️⃣ option to create a new Queryset.
#2. Name the Queryset Silver Layer rt_kqlqs_create-silver-layer
#3. Let's rename the default tab that we are working to Create Tables 1️⃣, copy the Code below and paste it into the Queryset window 2️⃣, select all the code and click Run! 3️⃣. The KQL will define the column names and data types for the table and stored it in the "silver" folder.
//Create 'OperativeStations' table attributes in SILVER LAYER folder
.execute database script <|
// Contains all orders for Rapid Delivery only.
.create table [OperativeStations] (
BikepointID: string,
Neighbourhood: string,
Street: string,
Latitude: real,
Longitude: real,
No_Bikes: long ,
No_Empty_Docks: long,
TotalCapacity: long,
BikeUsageRate: real,
Region: string,
IsCentral: int,
StatusName: string,
StatusKey: int,
EventEnqueuedUtcTime: datetime,
UpdatedAt: datetime
) with (folder = "silver")
Task 2: Initial load of SilverRapidOrders table
In the next step, you will create an update policy to keep your silver layer up to date as new data is received. However, we first want to backfill our tables with historical data. In this section you will use the .append command to perform the backfill operation.
🤓 The .append command in KQL is a one time operation that is used to add data to an existing table. When you use the .append command, it takes the results of a query and appends them to the specified table. This is particularly useful in our example to backfill the table with historical data.
#1. Create a new tab in your rt_kqlqs_create-silver-layer queryset by clicking on the plus sign ➕, click the pencil icon on the new tab and name it Append Data 1️⃣, copy the code below and paste it into the tab 2️⃣, select all and Run 3️⃣.
//Append data from 'in-operation', 'stations', and 'station-status' from 'bronze' folder into the 'OperativeStations' silver table
.execute database script <|
.append ['OperativeStations'] <|
['in-operation']
| join kind = inner (['stations']) on ($left.BikepointID == $right.BikepointId and $left.Neighbourhood == $right.Neighbourhood)
| extend StatusKey = toint(1)
| join kind = inner (['station-status'] | project StatusName = Name, StatusKey) on StatusKey
| extend UpdatedAt = now() // Assuming same time on first insert
// Update Data Types on existing columns
| summarize arg_max(EventEnqueuedUtcTime,*) by BikepointID
| project BikepointID,Neighbourhood,Street,Latitude,Longitude,No_Bikes,No_Empty_Docks,TotalCapacity,BikeUsageRate,Region,IsCentral,StatusName,StatusKey,EventEnqueuedUtcTime,UpdatedAt
#2. Once you run the KQL statement, you will see data immediately appear in your table. Running the following KQL query (perhaps in a new queryset tab) will help you validate your results, your results will be much different than the screenshot below, but you will have data!
///Number of rows in the 'OperativeStations' silver table
['OperativeStations']
| count
Task 3: Create Update Policies to Load the Silver Layer
Update Polices are very important when it comes to real-time intelligence
Update Policies rely on functions. Functions encapsulate the logic needed to transform and enrich data. Therefore, the process for using update policies is to first create a function with the transformational logic necessary to process the incoming data. The next step in the process is to set up an update policy on the table to use the function. The KQL functions and Alter table commands have been written for you in this section, however, take a moment to read through the KQL query so that you can better understand the process.
#1. Create a new tab in your Silver Layer queryset by clicking on the plus ➕, click the pencil icon on the new tab and name it Add policy, copy the code below and paste it into the tab 2️⃣, select all and Run 3️⃣.
// Use update policies to transform data during Ingestion
.execute database script <|
.create-or-alter function with (docstring = 'Add additional columns and data enrichment') fn_OperativeStations (){
['in-operation']
| join kind = inner (['stations']) on ($left.BikepointID == $right.BikepointId and $left.Neighbourhood == $right.Neighbourhood)
| extend StatusKey = toint(1)
| join kind = inner (['station-status'] | project StatusName = Name, StatusKey) on StatusKey
| extend UpdatedAt = now() // Assuming same time on first insert
// Update Data Types on existing columns
| summarize arg_max(EventEnqueuedUtcTime,*) by BikepointID
| project BikepointID,Neighbourhood,Street,Latitude,Longitude,No_Bikes,No_Empty_Docks,TotalCapacity,BikeUsageRate,Region,IsCentral,StatusName,StatusKey,EventEnqueuedUtcTime,UpdatedAt
}
// Create an update policy on the silver table
.alter table ['in-operation'] policy streamingingestion disable
.alter table ['OperativeStations'] policy update @'[{"Source": "in-operation", "Query": "fn_OperativeStations", "IsEnabled" : true, "IsTransactional": true }]'
What is the KQL Code in this update policy doing?
The KQL code used in this section will perform two primary operations.
- First, we create a function for each update policy.
- Second, we alter the silver table and add an update policy using the function.
🤓 Functions in KQL are crucial because they define the specific transformation logic applied to the data. These functions are used for data transformation, data enrichment and automation.
fn_OperativeStations
The function fn_OperativeStations in your script performs data transformation and enrichment during streaming ingestion into the OperativeStations table. It's used as part of an update policy, which means it's automatically triggered every time new data is ingested into the source table in-operation.
The Update Policies
An .alter table command is applied to each of the silver tables to add an update policy. The update policy will monitor the base source table and retrieve new records from the source through ingestion batches. The update policy will then call the defined function to enrich the data. Finally, the new data, will be inserted into the tables, respectively. It is important to note that these update policies are only inserting new rows of data, there is no deletion or updating occurring in the KQL Script below.
#2. Within a few moments of enabling your update policy you will begin to see new records flowing into your table, it was a 100+ records minutes ago!
#3. Validate policies using the .show command to view the current update policy on a table, run the script below in a new query tab
//use the .show command to view the current update policy on a table.
.show table OperativeStations policy update
Alerts with Activator
Task 4: Creating Alerts with Activator
Now that we have established our enriched silver layer, we can create alerts to detect abnormal or critical operational states in bike rental stations — specifically, when a station is completely full (i.e. has no empty docks) or like in this case, stations that have no additional bicycles for rent. To achieve this we can leverage Activator which is a no-code experience for automatically taking actions when patterns or conditions are detected in streaming data.
There are various ways to create an activator, either through a tile on a real-time dashboard, creating one from the + New item menu in the workspace and pointing to a stream of data, or from a KQL Queryset which is what you will be doing in this lab.
#1. Navigate to the KQL Database you have been using within your Fabric workspace 1️⃣, from the menu bar at the top select the KQL Queryset 2️⃣ option to create a new Queryset.
#2. Name the new KQL Queryset rt_kqlqs_activator-query and hit create.
#3. Delete any existing KQL code that may exist in the default query tab and re-label the tab to Empty Stations 1️⃣, copy the code below and paste it into the tab 2️⃣, select all and Run 3️⃣, most likely you will see the query will return the stations with no bicycles for rent 4️⃣, click on Set alert 5️⃣
//Identifies stations that have no bikes for rent
OperativeStations
| summarize arg_max(UpdatedAt,*) by BikepointID
| project BikepointID,Neighbourhood,UpdatedAt,No_Bikes,No_Empty_Docks,TotalCapacity,BikeUsageRate,IsCentral
| where UpdatedAt >= ago(30m) and BikeUsageRate >= 100
#4. Within the Set alert options we will choose to run the query every 5 minutes, leave the Condition to the default and only option, set the Action to Send me an email 1️⃣, and choose Create a new item and give the name act_alert_empty-operative-stations 2️⃣3️⃣ and Create 4️⃣
#5. Once the process is complete for creating the alert you will have an option available called Open which will open a new tab displaying the Activator user interface... a dialogue will appear giving you the opportunity to guide you through this user interface, feel free to explore this if you like but we will be selecting the Get started option to move forward. However, opening the new tab could also change your tenant - in this case, change your tenant back to 'Accenture Norway EA' and manually select the Activator item, act_alert_empty-operative-stations from your workspace.
#6. In this activator you have a single stream of data which is an event based on the query we just created 1️⃣, we can also see that we have a single rule, click to update with some dynamic content 2️⃣
#7. There are various areas of interest when looking at a rule, but the one you will focus on is labeled Definition which allows you to configure what you are monitoring 1️⃣, what is the condition which will trigger the action 2️⃣, and what action will be taken when the condition is met 3️⃣.
#8. Now you will configure the Subject with Empty station detected!1️⃣, Headline with An empty station has been detected in @Neighbourhood 2️⃣ and Message with The station @BikepointID located in @Neighbourhood with a total capacity of @TotalCapacity is empty, dispatch refilling truck 3️⃣ (you might need to type a space after each variable (@Neighbourhood e.g.) and select it from the drop-down menu, for it to register as an actual variable for the alert service).
#9. The last property that you will configure is the Context, currently Activator's are limited to only allowing 5 distinct columns to be used in the action, choose BikepointId, Neighborhood, TotalCapacity and IsCentral and click Save and Start.
#10. We have now finished configuring our first Activator. To see what the alert will look like choose the Send me a test action button near the bottom, check your email... you should see something like this 😉👇
#11. Click Stop to deactivate the rule and prevent spam on your inbox
Lab 4 Wrap‑Up: What You’ve Learned
In this lab, you advanced your real‑time data pipeline by building the Silver layer of your Medallion architecture and automating data updates and alerts:
- Created and populated Silver‑layer tables with enriched data.
- Performed an initial load to backfill data from the Bronze layer and defined Update Policies and functions that automatically transform and move streaming data into your Silver tables without manual intervention.
- Set up an Activator alert to detect and notify when bike stations are full or otherwise operationally critical, integrating real‑time detection with action.
➡️ Next up: Lab 5 – Real‑Time Dashboards, where you’ll visualize these insights and deliver them to decision‑makers.
