Performing ETL in Real-Time Intelligence with Microsoft Fabric

In today’s data-driven world, the ability to act on data as soon as it is generated is critical to helping businesses quickly make informed decisions. Organizations want to leverage the power of modern data to drive operations, marketing strategies, and customer interactions.

This becomes challenging in the world of real-time data, where it is not always possible to perform all transformations while the data is streaming. So you have to come up with a fast flow without affecting the data stream.

This is where Microsoft Fabric comes into play. Fabric offers a comprehensive suite of services including data engineering, data factory, data science, real-time intelligence, data warehouse, and database. But today we will focus on Real-Time Intelligence.

Use cases

This setting can be used in scenarios where data needs to be transformed for use in downstream processing/analytics workloads. An example of this is: OneLake Availability You can use that data in KQL tables to access it from other Fabric engines like Notebooks, Lakehouse, etc. for training ML models/analysis.

As another example, let’s say you have a timestamp column in your streaming data and you want to change its format according to a standard. You can use update policies to convert and store timestamp data formats.

Fabric Real-Time Intelligence supports the KQL database as a data store designed to efficiently process real-time data streams. You can use it after ingestion. Kusto Query Language (KQL) Query data from a database.

A KQL table is a fabric item that is part of the KQL database. Both of these entities are Event House. Eventhouse is a workspace in a database that can be shared across specific projects. This allows you to manage multiple databases simultaneously and share capacity and resources to optimize performance and costs. Event House provides integrated monitoring and management for all databases and for each database.

Figure 1: Fabric Item Hierarchy in Eventhouse

Policy updates This is an automated process that becomes active when new data is added to the table. It automatically transforms data coming in through queries and stores the results in the target table, eliminating the need for manual reconciliation. A single table can have multiple update policies for different transformations, allowing data to be stored in multiple tables simultaneously. These target tables can have a separate schema, retention policy, and other configuration from the source table.

This blog contains a scenario where data enrichment is performed on data in a KQL table. In this case, you delete columns that you don’t need, but you can also perform other transformations supported by KQL on the data.

Here we have a real-time stream that pushes data into a KQL table. Once the source table is loaded, we drop unneeded columns and use an update policy to push the data of interest from the source table to the target table.

Create a sample data stream

1. In real-time intelligence experience, Create a new event stream.
2. Add a new source under Sources and select Sample Data.

   

3. Continue configuring the stream. I’m using the Bicycles sample data stream in this blog.
4. Select Direct collection as the data collection mode for the target.
5. For Target, select the workspace and KQL database that you created as prerequisites for this exercise.
6. You should see a pop-up to configure your database details and continue configuring the tables through which data should be passed from the stream.

Configure KQL table with update policy

1. Open the Eventhouse page in Fabric. You can now preview the data being collected in the sample data stream.

2. Create a new target table. I created a new table (target) using the following KQL:

   .create table RTITableNew (
       BikepointID: string,Street: string, Neighbourhood: string, No_Bikes: int, No_Empty_Docks: int )
   

3. On the Database tab, click New and select Table Update Policy.

   

4. You can edit an existing policy format or paste the format I used below.

   Note: RTITable is the source and RTITableNew is the target table.

   .alter table RTITable policy update ```[
     {
       "IsEnabled": true,
       "Source": "RTITable",
       "Query": "RTITable | project BikepointID=BikepointID, Street=Street, Neighbourhood=Neighbourhood, No_Bikes=No_Bikes, No_Empty_Docks=No_Empty_Docks ",
       "IsTransactional": true,
       "PropagateIngestionProperties": false,
       "ManagedIdentity": null
     }
   ]```​

   The above policy will drop the longitude and latitude columns and store the remaining columns in the target table. Depending on your requirements, you can perform more conversions, but the workflow remains the same.

5. After running the above command, the target table will start populating with new data as soon as the source table gets its data. To review the policy of the target table, you can run the following command:

    .show table  policy update​

In summary, we took a real-time data stream, stored the data in a KQL database, then performed data enrichment on the data and stored it in a destination table. This flow satisfies any scenario where you want to perform processing on data collected from a stream.

Common scenarios for using table update policies – Kusto | microsoft run

Creating table update policies in Real-Time Intelligence – Microsoft Fabric | microsoft run