Streaming Data into a Fabric Eventhouse with Eventstreams: Configuring Sources, Defining KQL Tables, and Querying Live Telemetry Data

Real-time data is different in kind, not just in speed. When a temperature sensor fires an alert, a payment fraud signal fires, or a fleet vehicle crosses a geofence, you can't afford to wait for the next batch job to finish. You need to ingest, store, and query that event within seconds — and you need the query engine to stay fast even as millions of rows accumulate in the background. That's the gap that Microsoft Fabric's real-time analytics stack was purpose-built to close.

In this lesson, you'll build a working real-time telemetry pipeline from end to end. We'll connect a streaming source to an Eventstream, route that stream into an Eventhouse (Fabric's managed KQL database service), define tables that receive the data continuously, and write KQL queries that answer questions about live events. The scenario we'll use throughout is an IoT fleet-monitoring system: dozens of delivery trucks sending GPS coordinates, engine temperature, and fuel level every 30 seconds. This is exactly the kind of workload that breaks batch pipelines and thrives on streaming architecture.

What you'll learn:

• How Eventstreams work as a managed streaming pipeline and which source types they support
• How to configure an Eventhouse and create a KQL database to receive streaming data
• How to map an Eventstream output to a KQL table with the right ingestion mapping
• How to write KQL queries that operate against live, continuously-updating data
• How to use update policies and materialized views to transform and pre-aggregate telemetry at ingestion time

---

Prerequisites

You should already be comfortable with the basics of Microsoft Fabric before diving in here. Specifically:

• You have an active Fabric capacity (F2 or higher, or a trial) and a workspace configured. If you need a refresher, see Fabric Capacities and Workspaces: F SKUs, Trials, and Setting Up Your First Workspace.
• You understand what OneLake is and how Fabric items relate to each other — What Is Microsoft Fabric? Workloads, OneLake, and How It Fits with Power BI covers this well.
• You've at least skimmed the conceptual overview of Real-Time Analytics in Microsoft Fabric: Eventstreams, Eventhouses, and KQL Querysets — this lesson builds on those concepts with hands-on implementation.
• You have some familiarity with JSON as a data format. All IoT telemetry in this lesson arrives as JSON.

---

The Architecture You're Building

Before touching any UI, it's worth being precise about what each component does and why you need it.

Eventstream is Fabric's managed streaming pipeline. Think of it as a no-code/low-code version of Azure Stream Analytics fused with Event Hubs routing logic. You point it at a source (an Event Hub, IoT Hub, custom application, or Fabric-native sample data), optionally transform the data mid-stream, and route it to one or more destinations. Eventstream handles the underlying Kafka-compatible infrastructure, checkpointing, and at-least-once delivery guarantees so you don't have to.

Eventhouse is Fabric's container for one or more KQL databases. Each KQL database runs on the Kusto engine — a columnar, append-optimized store designed specifically for time-series and log data. Eventhouse is where your events land and live. Unlike a Fabric warehouse or lakehouse, an Eventhouse is not backed by Delta Lake files in OneLake. It maintains its own columnar storage optimized for the append-then-query pattern that telemetry demands. (You can export data from an Eventhouse to OneLake as a shortcut for cross-workload querying, which we'll touch on later.)

KQL (Kusto Query Language) is the query language you use against an Eventhouse. If you've written SQL, KQL will feel familiar in structure but different in syntax. It's pipeline-oriented — you start with a table and pipe the result through a series of operators. KQL is also the language you use to define schemas, ingestion mappings, and materialized views inside the database itself.

The flow for our fleet monitoring scenario looks like this:

1. Truck telemetry is published to an Azure Event Hub (one event per vehicle per 30-second interval).
2. An Eventstream ingests from that Event Hub and routes to an Eventhouse KQL database.
3. A raw table TruckTelemetry_Raw receives every event.
4. An update policy transforms and fans events into a TruckTelemetry table with typed columns.
5. A materialized view TruckTelemetry_Hourly pre-aggregates average speed and temperature by truck and hour.
6. A KQL queryset answers operational questions: which trucks are running hot, which are behind schedule, where is vehicle 47 right now.

---

Step 1: Create an Eventhouse and KQL Database

Open your Fabric workspace. In the left-hand nav, select New item, then scroll to the Real-Time Intelligence section and choose Eventhouse. Give it a name — FleetMonitoring works. Click Create.

Fabric automatically creates one KQL database inside the Eventhouse with the same name. You'll see the Eventhouse landing page with a database tile showing storage stats (both are zero at this point — that's fine).

Click into the KQL database. You'll land on the database home page, which shows three key things: Shortcuts, Tables, and Query shortcuts. The most important button on this page is the pencil icon that opens the KQL query editor — this is where you'll define your schema and write your queries.

Define the Raw Table

The first table you create is intentionally wide-open. Instead of defining a strict schema up front, you'll accept the raw JSON payload as a single dynamic column and let a transformation policy parse it. This pattern — raw table plus update policy — is idiomatic KQL and gives you a reprocessing escape hatch: if your transformation logic has a bug, the raw events are still there.

Open the KQL query editor and run:

// Create the raw landing table
.create table TruckTelemetry_Raw (payload: dynamic)

// Set 7-day retention on the raw table — we don't need to keep raw forever
.alter table TruckTelemetry_Raw policy retention

{
  "SoftDeletePeriod": "7.00:00:00",
  "Recoverability": "Disabled"
}

Then define the ingestion mapping. This tells the Kusto engine how to map an incoming JSON event to the payload column:

.create table TruckTelemetry_Raw ingestion json mapping "RawMapping"
'[{"column":"payload","Properties":{"Path":"$"}}]'

The $ path means "take the entire JSON document and put it in the payload column." Simple, but powerful.

Define the Typed Table

Now create the table that will actually hold structured, queryable data:

.create table TruckTelemetry (
    EventTime: datetime,
    TruckId: string,
    Latitude: real,
    Longitude: real,
    SpeedKph: real,
    EngineTemp_C: real,
    FuelLevel_Pct: real,
    RouteId: string,
    DriverId: string
)

This is the table you'll query most of the time. The types matter: datetime for timestamps gives you access to KQL's rich time-series operators, real for numeric measurements, string for categorical fields.

Define the Update Policy

The update policy is a KQL function that Kusto runs automatically on every batch of rows inserted into TruckTelemetry_Raw. The output is appended into TruckTelemetry. Think of it as a persistent, streaming ETL transformation that requires no external scheduler.

// Define the transformation function
.create function TruckTelemetry_Transform() {
    TruckTelemetry_Raw
    | extend 
        EventTime     = todatetime(payload.timestamp),
        TruckId       = tostring(payload.truck_id),
        Latitude      = todouble(payload.gps.lat),
        Longitude     = todouble(payload.gps.lon),
        SpeedKph      = todouble(payload.speed_kph),
        EngineTemp_C  = todouble(payload.engine_temp_c),
        FuelLevel_Pct = todouble(payload.fuel_level_pct),
        RouteId       = tostring(payload.route_id),
        DriverId      = tostring(payload.driver_id)
    | project EventTime, TruckId, Latitude, Longitude, 
              SpeedKph, EngineTemp_C, FuelLevel_Pct, RouteId, DriverId
}

// Attach the update policy to TruckTelemetry_Raw
.alter table TruckTelemetry policy update
@'[{"IsEnabled": true, "Source": "TruckTelemetry_Raw", "Query": "TruckTelemetry_Transform()", "IsTransactional": false, "PropagateIngestionProperties": false}]'

The IsTransactional: false setting means that if the transformation function fails for a batch, the raw data is still committed to TruckTelemetry_Raw. Setting it to true would roll back both tables on failure, which is safer but slower. For telemetry at high throughput, false is almost always the right choice — you'd rather lose a few transformed rows than block the entire ingestion pipeline.

---

Step 2: Configure the Eventstream

Go back to your workspace, create a New item, and choose Eventstream. Name it FleetTelemetry_Stream.

Connecting a Source

The Eventstream canvas opens with an empty pipeline. Click Add source in the toolbar. You'll see a list of available source types:

• Azure Event Hubs — the most common production source
• Azure IoT Hub — for devices registered in Azure IoT
• Custom App — generates connection strings for your own producers
• Azure Service Bus — for queue or topic-based producers
• Sample data — Fabric-provided synthetic streams for learning

For production, you'd choose Event Hubs and supply your namespace, hub name, consumer group, and a connection string or managed identity credential. For this walkthrough, let's use Sample data first to validate the pipeline before connecting real hardware. Select it, choose the Bicycles sample (it provides a realistic time-series stream), and name the source node SampleTelemetry.

When you're ready to connect a real Event Hub, the configuration fields are:

• Event Hub namespace: your Azure namespace FQDN (e.g., fleetevents.servicebus.windows.net)
• Event Hub name: the specific hub within the namespace
• Consumer group: use a dedicated consumer group ($fabric-eventstream) — never $Default in production, since other consumers sharing $Default can interfere with offset tracking
• Connection: choose "New connection" and either paste a SAS key or configure managed identity

Adding a Destination

With the source node on the canvas, click Add destination and choose KQL Database. A destination node appears — drag a connector from your source node to it.

Click the destination node to open its configuration panel on the right:

• Workspace: select your workspace
• Eventhouse: FleetMonitoring
• KQL Database: FleetMonitoring
• Table: TruckTelemetry_Raw — this is the raw landing table you defined
• Input data format: JSON
• Mapping name: RawMapping — the ingestion mapping you created earlier

Click Save on the destination configuration. Then click Publish in the Eventstream toolbar. Publishing activates the stream — data starts flowing within about 30 seconds.

Adding a Transformation (Optional but Powerful)

Between your source and destination nodes, you can insert transformation operators. Click Add operation on the connector between source and destination to insert nodes for:

• Filter: drop events that don't match a condition (e.g., filter out events where speed_kph == 0 to reduce storage)
• Aggregate: compute running windowed aggregates (tumbling, hopping, or session windows)
• Expand: flatten nested JSON arrays into individual events
• Group by: partition the stream by a key before aggregating

For the fleet scenario, a useful transformation is a Filter that drops heartbeat pings sent while trucks are parked at the depot, identified by a status == "parked" field. Filtering upstream in the Eventstream reduces ingestion volume and cost before data touches the Eventhouse.

---

Step 3: Verify Ingestion Is Working

Give the pipeline a minute to run after publishing, then go back to your KQL database and open the query editor. Run:

// Check row counts in both tables
TruckTelemetry_Raw
| count

TruckTelemetry
| count

If TruckTelemetry_Raw has rows but TruckTelemetry has zero, your update policy isn't firing correctly. The most common cause is a type conversion failure in the transform function — todouble() on a null or non-numeric field will produce null, not an error, but todatetime() on a malformed timestamp string can silently discard rows. We'll cover debugging patterns in the troubleshooting section.

A quick sanity check on the data:

TruckTelemetry
| take 10
| project EventTime, TruckId, SpeedKph, EngineTemp_C, FuelLevel_Pct
| order by EventTime desc

You want to see recent timestamps in EventTime, sensible numeric values, and non-null truck IDs.

---

Step 4: Build a Materialized View for Aggregated Telemetry

Materialized views are one of the most powerful features in the Kusto engine, and they're especially well-suited to telemetry workloads. A materialized view computes an aggregation incrementally as new data arrives and stores the result — so when you query the view, Kusto answers from pre-computed results rather than scanning millions of raw rows.

.create materialized-view with (backfill=true) TruckTelemetry_Hourly on table TruckTelemetry
{
    TruckTelemetry
    | summarize 
        AvgSpeed        = avg(SpeedKph),
        MaxEngineTemp   = max(EngineTemp_C),
        MinFuelLevel    = min(FuelLevel_Pct),
        EventCount      = count()
        by TruckId, RouteId, bin(EventTime, 1h)
}

The backfill=true option tells Kusto to process data that was already in the table when you created the view — without it, the view only materializes new data going forward.

You can also create a "latest record per truck" view, which is extremely useful for a live vehicle dashboard:

.create materialized-view TruckTelemetry_Latest on table TruckTelemetry
{
    TruckTelemetry
    | summarize arg_max(EventTime, *) by TruckId
}

arg_max(EventTime, *) returns the row with the maximum EventTime value for each TruckId, along with all other columns (*). This gives you the most recent known position, speed, and status for every truck — updated continuously as new telemetry arrives.

---

Step 5: Write Operational KQL Queries

This is where real-time analytics pays off. Let's write the queries an operations team would actually run.

Which Trucks Are Running Hot Right Now?

TruckTelemetry_Latest
| where EngineTemp_C > 105
| project TruckId, DriverId, RouteId, EngineTemp_C, SpeedKph, EventTime
| order by EngineTemp_C desc

Because TruckTelemetry_Latest is a materialized view, this query is very fast — it scans only one row per truck rather than the full history. If you need the threshold to be dynamic (loaded from a reference table), you can join:

let alert_threshold = 105.0;
TruckTelemetry_Latest
| where EngineTemp_C > alert_threshold
| join kind=leftouter (
    RouteDefinitions  // hypothetical lookup table
    | project RouteId, RouteDescription, DispatchCenter
) on RouteId
| project TruckId, DriverId, RouteDescription, DispatchCenter, 
          EngineTemp_C, EventTime

Speed Trend for a Specific Truck Over the Last Hour

let truck_id = "TRK-047";
let lookback = 1h;
TruckTelemetry
| where TruckId == truck_id
    and EventTime >= ago(lookback)
| project EventTime, SpeedKph, EngineTemp_C, FuelLevel_Pct
| order by EventTime asc
| render timechart

The render timechart directive renders a line chart directly in the KQL query editor — useful for ad-hoc investigation without needing to open Power BI. In production, you'd remove render and connect the query to a Power BI semantic model or a Real-Time Dashboard.

Hourly Summary by Route for the Last 24 Hours

TruckTelemetry_Hourly
| where EventTime >= ago(24h)
| summarize 
    AvgSpeed      = avg(AvgSpeed),
    MaxEngineTemp = max(MaxEngineTemp),
    TruckCount    = dcount(TruckId)
    by RouteId, bin(EventTime, 1h)
| order by RouteId asc, EventTime asc

Notice we're querying the materialized view (TruckTelemetry_Hourly) and summarizing again — this is perfectly valid. Kusto folds the second aggregation on top of the pre-computed view efficiently.

Detect Anomalous Engine Temperature Using Series Analysis

KQL has built-in time-series anomaly detection. This is one of those features that would take weeks to build manually:

let truck_id = "TRK-023";
TruckTelemetry
| where TruckId == truck_id
    and EventTime >= ago(6h)
| make-series AvgTemp = avg(EngineTemp_C) on EventTime 
    from ago(6h) to now() step 2m
| extend anomalies = series_decompose_anomalies(AvgTemp, 2.5)
| mv-expand EventTime, AvgTemp, anomalies
| where tolong(anomalies) != 0
| project EventTime, AvgTemp, anomalies

series_decompose_anomalies runs a decomposition model that separates trend, seasonality, and residual noise, then flags points where the residual exceeds a threshold (2.5 standard deviations here). In a fleet context, this catches a truck whose engine is overheating gradually — not yet over the hard threshold, but trending toward it.

---

Step 6: Expose Data to the Broader Fabric Ecosystem

An Eventhouse doesn't live in isolation. Fabric lets you surface KQL data to other workloads.

Creating a KQL Queryset

A KQL Queryset is a saved, shareable set of KQL queries — like a notebook, but purpose-built for the Kusto engine. Create one from your workspace (New item → KQL Queryset) and connect it to your FleetMonitoring database. Save your operational queries here so your team can access and modify them without touching the raw schema objects.

Surfacing an Eventhouse in Power BI

You can connect Power BI directly to a KQL database using the Azure Data Explorer (Kusto) connector. Point it at your Eventhouse's query URI (visible in the Eventhouse settings page, formatted as https://<eventhouse-name>.<region>.kusto.fabric.microsoft.com). Use a KQL query as the data source query for each table in your semantic model.

The caveat: this uses DirectQuery mode, which means every Power BI visual fires a KQL query at render time. For a real-time dashboard with a 30-second auto-refresh, this works well. For complex reports with many visuals and slicers, the query fan-out can be substantial — consider whether materialized views or pre-aggregated tables can reduce per-visual query cost. This is a different trade-off from Direct Lake Mode in Power BI, which applies to Delta tables in OneLake, not KQL databases.

Exporting to OneLake

If your downstream workloads — say, a Spark notebook running a machine learning model — need access to the telemetry data, you can configure a continuous export from the KQL database to OneLake as Parquet files:

.create-or-alter continuous-export TruckTelemetry_Export
over (TruckTelemetry)
to table FleetLakehouse.TruckTelemetry_Export
with (intervalBetweenRuns=5m, forcedLatency=10m)
{
    TruckTelemetry
    | where EventTime >= ago(1d)
}

This makes the telemetry available as Delta-compatible Parquet in a lakehouse, bridging the real-time and batch worlds without duplicating the streaming pipeline. The lakehouse then participates in the broader OneLake ecosystem.

---

Hands-On Exercise: Build a Live Fuel Monitoring Pipeline

Now put everything together. You'll build a complete pipeline that monitors fuel levels across the fleet and identifies trucks that need refueling before their next stop.

Step 1: Create a fresh KQL database called FuelMonitoring inside your existing FleetMonitoring Eventhouse.

Step 2: Define the schema. Run these commands in the KQL editor:

// Raw landing table
.create table FuelEvents_Raw (payload: dynamic)

.create table FuelEvents_Raw ingestion json mapping "FuelMapping"
'[{"column":"payload","Properties":{"Path":"$"}}]'

// Typed table
.create table FuelEvents (
    EventTime:       datetime,
    TruckId:         string,
    FuelLevel_Pct:   real,
    FuelRate_LPH:    real,      // litres per hour consumption rate
    DistanceToDepot: real,      // km
    RouteId:         string
)

// Transform function
.create function FuelEvents_Transform() {
    FuelEvents_Raw
    | extend
        EventTime       = todatetime(payload.timestamp),
        TruckId         = tostring(payload.truck_id),
        FuelLevel_Pct   = todouble(payload.fuel_level_pct),
        FuelRate_LPH    = todouble(payload.fuel_rate_lph),
        DistanceToDepot = todouble(payload.distance_to_depot_km),
        RouteId         = tostring(payload.route_id)
    | project EventTime, TruckId, FuelLevel_Pct, FuelRate_LPH, DistanceToDepot, RouteId
}

// Update policy
.alter table FuelEvents policy update
@'[{"IsEnabled": true, "Source": "FuelEvents_Raw", "Query": "FuelEvents_Transform()", "IsTransactional": false, "PropagateIngestionProperties": false}]'

// Latest-per-truck view
.create materialized-view FuelEvents_Latest on table FuelEvents
{
    FuelEvents
    | summarize arg_max(EventTime, *) by TruckId
}

Step 3: Create an Eventstream called FuelStream. Connect the Sample data source (or your Event Hub). Route to FuelEvents_Raw in the FuelMonitoring database with the FuelMapping mapping. Publish it.

Step 4: After a minute, run this query to identify trucks that will run out of fuel before reaching the depot:

// Estimate remaining range in km, assuming 30L tank capacity
let tank_capacity_L = 30.0;
let avg_fuel_economy = 0.12;  // litres per km

FuelEvents_Latest
| extend
    FuelRemaining_L  = FuelLevel_Pct / 100.0 * tank_capacity_L,
    EstimatedRange_km = (FuelLevel_Pct / 100.0 * tank_capacity_L) / avg_fuel_economy
| where EstimatedRange_km < DistanceToDepot * 1.15   // less than 15% buffer
| project TruckId, RouteId, FuelLevel_Pct, EstimatedRange_km, 
          DistanceToDepot, EventTime
| order by EstimatedRange_km asc

Step 5: Create a KQL Queryset named FuelOpsQueries and save this query plus a fuel trend query for the last 4 hours. Share it with a colleague by assigning them Viewer access on the KQL Queryset item.

---

Common Mistakes & Troubleshooting

"TruckTelemetry is empty but TruckTelemetry_Raw has rows"

This almost always means the update policy function is failing silently. Run the transform function manually against recent raw data to see what's happening:

TruckTelemetry_Raw
| take 100
| extend 
    EventTime = todatetime(payload.timestamp),
    TruckId   = tostring(payload.truck_id)
| where isnull(EventTime) or isempty(TruckId)

If this returns rows, your JSON paths are wrong or the timestamp format doesn't parse. Check the actual payload structure with:

TruckTelemetry_Raw
| take 1
| project payload

Then expand the JSON to see the actual field names. Pay attention to case — payload.Timestamp and payload.timestamp are different in KQL.

"Eventstream shows data flowing but nothing arrives in the table"

Check the destination node in the Eventstream canvas. Click the destination node and look at the Operational metrics tab — you'll see ingestion error counts and the last successful ingestion time. Ingestion mapping mismatches (wrong mapping name, or mapping was deleted and recreated) are the top cause.

Also verify the consumer group. If $Default is being used and another process is consuming from the same Event Hub with that group, Eventstream may see no events because the other consumer is reading them first.

"Query is slow on recent data"

The most likely cause is a hot cache miss. The default caching policy keeps 31 days in hot cache, but if your query time range extends beyond that (e.g., where EventTime >= ago(60d)), Kusto fetches from cold storage, which is dramatically slower.

Check your caching policy:

.show table TruckTelemetry policy caching

Extend it if your queries need a wider hot window:

.alter table TruckTelemetry policy caching hot = 90d

"Materialized view is lagging by more than 10 minutes"

Materialized views have a background job that processes new data. Under high ingestion throughput, this job can fall behind. First, check the view's lag:

.show materialized-view TruckTelemetry_Hourly

Look at the Age column — this is the lag. If it's consistently above a few minutes, the view's complexity or your ingestion rate may be too high for the default background job frequency. In extreme cases, consider simplifying the aggregation or sharding the view across multiple tables by truck region.

"Update policy function references a column that doesn't exist in the source"

KQL functions in update policies are compiled at creation time but may fail at runtime if the data doesn't match expectations. Always wrap nullable fields in defensive type conversions:

// Defensive version — won't fail on null or missing fields
FuelRate_LPH = coalesce(todouble(payload.fuel_rate_lph), 0.0)

coalesce returns the first non-null value — pairing it with todouble() means that a missing or malformed field produces 0.0 instead of null, which prevents type-conversion failures from silently dropping rows.

---

Summary & Next Steps

You've built a complete real-time telemetry pipeline in Microsoft Fabric: from a streaming source through an Eventstream, into an Eventhouse with a raw landing table, through an update policy transformation, and out through materialized views that power operational queries. The KQL you've written covers the full range from simple spot checks to time-series anomaly detection — the kind of analysis that was either impossible or prohibitively expensive to run in batch systems.

The key architectural principles to carry forward:

• Raw table + update policy is idiomatic KQL. Never ingest directly into your typed table if you can avoid it — the raw table is your recovery mechanism.
• Materialized views are your performance lever. Build them for any aggregation pattern that repeats across multiple queries.
• EventTime vs. ingestion_time() — always track both, always query on EventTime.
• Hot cache policy is your main cost dial. Set it to cover your most common query patterns, not your longest possible query range.

For your next steps, consider exploring how to orchestrate batch loads alongside your streaming pipeline — Orchestrating Loads with Fabric Data Pipelines: Copy Activities, Parameters, and Schedules shows you how to build the batch side of a lambda architecture. If your operations team needs to govern who can query which telemetry, Securing and Governing Microsoft Fabric: Workspace Roles, Item Permissions, and OneLake Data Access walks through access control at the item and column level. And if you want to surface your KQL materialized views in a Power BI report with live refresh, the patterns in Real-Time Analytics in Microsoft Fabric: Eventstreams, Eventhouses, and KQL Querysets cover the reporting integration in depth.

Streaming data is genuinely different from batch — the hardest part isn't the technology, it's shifting your intuition away from "run the job, check the output" and toward "the data is always flowing; what do I need to know right now?" You've built the infrastructure to answer that question.