In this post we are going to cover operations, monitoring and integration in Azure Data Explorer. These are 7 real Azure Data Explorer and KQL interview questions for data engineers, asked the way they come up on a real project, and the last one is a scenario: data stopped arriving in a table for two hours and nobody noticed.
Operations is everything after the cluster is built: knowing it's healthy, finding failed loads, reading the lake, connecting tools, exporting data, and shipping schema changes safely. Interviewers ask about it because this is the work of a running platform, and the questions show quickly who has been on call for one.
Part 9 of the Real Time Azure Data Explorer and KQL Interview Questions and Answers series on TechBrothersIT.
- The short answer
- Let's create some sample data
- Question 1: Your manager asks how you know the cluster is healthy. What do you monitor?
- Question 2: Someone asks how to find ingestion failures. Which commands do you use?
- Question 3: A teammate asks how to query data in the lake without ingesting it. How do external tables work?
- Question 4: Someone asks how Data Factory, Power BI and Grafana connect to Azure Data Explorer. What do you use?
- Question 5: A teammate needs query results exported to storage on a schedule. How does continuous export work?
- Question 6: Someone asks how to deploy tables, functions and policies safely across environments. What do you use?
- Scenario 7: Data stopped arriving in a table for two hours and nobody noticed. What do you put in place?
- The full script
The short answer
Watch Azure Monitor metrics (keep alive, CPU, cache utilization factor, ingestion latency and result) with alerts. .show ingestion failures explains failed loads, external tables query the lake in place, connectors send KQL to stored functions, continuous export uses database cursors, schema ships as idempotent scripts, and a freshness check per table catches silence.
Let's create some sample data
Every query below runs on the Kusto emulator, Microsoft's Azure Data Explorer engine in a local container, in a database called shop. kql is a small helper that sends a query or a .kql file to the emulator and prints the result as a table. setup.kql creates three small invented tables: Sales, six orders on October 20, Telemetry, three device readings, and an empty Orders table. Both loads use .ingest inline, which is for exploration and prototyping, not production. What only a cluster in Azure can show, such as Azure Monitor metrics and alerts, storage accounts and deployments, is shown as code.
.create table Sales (OrderId:long, Customer:string, Amount:real, OrderTime:datetime)
.ingest inline into table Sales <|
5001,C001,25.5,2026-10-20T08:05:00Z
5002,C002,14,2026-10-20T08:40:00Z
5003,C001,40,2026-10-20T09:15:00Z
5004,C003,9.5,2026-10-20T10:20:00Z
5005,C002,60,2026-10-20T10:55:00Z
5006,C004,33,2026-10-20T11:50:00Z
.create table Telemetry (DeviceId:string, Temp:real, Timestamp:datetime)
.ingest inline into table Telemetry <|
d1,21.5,2026-10-20T09:00:00Z
d2,22,2026-10-20T09:30:00Z
d1,21.8,2026-10-20T09:55:00Z
.create table Orders (OrderId:long, Customer:string, Amount:real)Question 1: Your manager asks how you know the cluster is healthy. What do you monitor?
Azure Monitor metrics, with alerts on the few that matter. Metrics tell you that something is wrong; diagnostic logs and the .show commands tell you why.
Let's go through them in order. Keep alive tells you the cluster answers queries. CPU and the cache utilization factor tell you about capacity: above 100 percent, the hot data no longer fits the cache. For ingestion, we watch ingestion latency, from received to ready for query, the ingestion result, and ingestion utilization. With data connections, discovery latency shows a backlog building before data is late. For queries, throttled queries and query duration.
| Metric | What it tells you | Alert when |
|---|---|---|
| Keep alive | the cluster answers queries | below 1 |
| CPU | compute pressure | high for a long time |
| Cache utilization factor | hot data vs cache (100% is optimal) | above 100% |
| Ingestion latency | received to ready for query | above your target |
| Ingestion result | successes and failures | any failures |
| Ingestion utilization | ingestion capacity in use | high for a long time |
| Discovery latency | data connection backlog | growing |
| Throttled queries | requests rejected by limits | above 0 |
And if you use them:
- Materialized views: age and health
- Continuous export: pending and lateness
- The why: diagnostic logs, .show commands
I alert on keep alive, ingestion failures, ingestion latency and throttling, routed to an action group. The emulator has no Azure Monitor, so this one is code:
# the cluster stops answering queries
az monitor metrics alert create -n adx-keepalive -g rg-data \
--scopes "$ADX_ID" --condition "avg KeepAlive < 1" \
--window-size 5m --evaluation-frequency 1m --action "$AG_ID"
# data takes longer than 10 minutes to become queryable
az monitor metrics alert create -n adx-ingest-latency -g rg-data \
--scopes "$ADX_ID" \
--condition "avg IngestionLatencyInSeconds > 600" \
--window-size 15m --evaluation-frequency 5m --action "$AG_ID"Alerting on CPU alone. A busy cluster is fine; late data is not.
Question 2: Someone asks how to find ingestion failures. Which commands do you use?
.show ingestion failures tells you why a load failed, with the failure kind and the error code. But first know this: a failed ingestion command doesn't always raise an error.
Let's prove it. We load three files into Orders with .ingest into: one has a short row, one is CSV loaded as Parquet, and one doesn't exist. Then we count the table.
.ingest into table Orders (@"/kustodata/orders_1001.csv")
with (format='csv', ignoreFirstRecord=true)
| project ItemLoaded, HasErrors
.ingest into table Orders (@"/kustodata/orders_1002.csv")
with (format='parquet')
| project ItemLoaded, HasErrors
.ingest into table Orders (@"/kustodata/orders_1003.csv")
with (format='csv')
| project ItemLoaded, HasErrors
Orders | countOrderId,Customer,Amount
6001,C004,12
6002,C0056003,C006,20
6004,C001,18$ kql -f ingest.kql
> .ingest into table Orders (@"/kustodata/orders_1001.csv")
ItemLoaded HasErrors
-------------------------- ---------
/kustodata/orders_1001.csv 1
> .ingest into table Orders (@"/kustodata/orders_1002.csv")
ItemLoaded HasErrors
-------------------------- ---------
/kustodata/orders_1002.csv 1
> .ingest into table Orders (@"/kustodata/orders_1003.csv")
ItemLoaded HasErrors
-------------------------- ---------
/kustodata/orders_1003.csv 1
> Orders | count
Count
-----
0No error. Each load returns HasErrors 1, which means true, and Orders has 0 rows.
Failures show up a little later on the emulator, so the full script waits at this point until all three are listed. Then we read them:
$ kql '.show ingestion failures | where Table == "Orders" | project Source = IngestionSourcePath, FailureKind, ErrorCode | order by Source asc'
Source FailureKind ErrorCode
-------------------------- ----------- --------------------------
/kustodata/orders_1001.csv Permanent Stream_WrongNumberOfFields
/kustodata/orders_1002.csv Permanent BadRequest_InvalidBlob
/kustodata/orders_1003.csv Permanent Download_SourceNotFound| Source | FailureKind | ErrorCode |
|---|---|---|
| /kustodata/orders_1001.csv | Permanent | Stream_WrongNumberOfFields |
| /kustodata/orders_1002.csv | Permanent | BadRequest_InvalidBlob |
| /kustodata/orders_1003.csv | Permanent | Download_SourceNotFound |
Each file is there, all Permanent: orders_1001.csv with Stream_WrongNumberOfFields, orders_1002.csv with BadRequest_InvalidBlob, and orders_1003.csv with Download_SourceNotFound.
Now the catch. This command covers only one stage:
| Where | Covers |
|---|---|
| .show ingestion failures | the management-command stage only, 14 days |
| Ingestion result metric + FailedIngestion log | every stage, queued and data connections too |
| .show operations <id> | the state of one async command |
A transient failure can work on a retry; a permanent one needs a fix to the data or the mapping. Also check OriginatesFromUpdatePolicy: the file can be fine while the update policy fails. For one async command, .show operations with its operation ID gives the state.
I summarize failures by error code per day and alert when a new error code shows up.
Question 3: A teammate asks how to query data in the lake without ingesting it. How do external tables work?
An external table is a schema over files in storage, like CSV, Parquet or Delta in ADLS, or over a SQL database. The data stays where it is, and you query it with external_table().
Let's create OrdersLake over two day folders. The partition column, Day, comes from the path format, not from the files, so a filter on Day reads only that folder. We query October 2.
.create external table OrdersLake (OrderId:long, Customer:string,
Amount:real)
kind=storage partition by (Day:datetime)
pathformat=('day=' datetime_pattern('yyyy-MM-dd', Day))
dataformat=csv
( h@'/kustodata/lake/orders' )
external_table('OrdersLake')
| where Day == datetime(2026-10-02)
| order by OrderId asc5001,C001,25.5
5002,C002,145003,C001,40
5004,C003,9.5
5005,C002,60$ kql -q -f lake.kql
> .create external table OrdersLake (OrderId:long, Customer:string,
ok: .create external table OrdersLake (OrderId:long, Customer:string,
> external_table('OrdersLake')
OrderId Customer Amount Day
------- -------- ------ --------------------
5003 C001 40 2026-10-02T00:00:00Z
5004 C003 9.5 2026-10-02T00:00:00Z
5005 C002 60 2026-10-02T00:00:00Z| OrderId | Customer | Amount | Day |
|---|---|---|---|
| 5003 | C001 | 40 | 2026-10-02T00:00:00Z |
| 5004 | C003 | 9.5 | 2026-10-02T00:00:00Z |
| 5005 | C002 | 60 | 2026-10-02T00:00:00Z |
October 2 returns its three orders, 5003, 5004 and 5005, each with Day 2026-10-02T00:00:00Z taken from the folder name.
Next, .show external table OrdersLake artifacts lists the files behind it, with the partition each one maps to.
$ kql '.show external table OrdersLake artifacts | project Uri, Partition | order by Uri asc'
Uri Partition
------------------------------------------------ --------------------------------------
/kustodata/lake/orders/day=2026-10-01/part-0.csv {"Day":"2026-10-01T00:00:00.0000000Z"}
/kustodata/lake/orders/day=2026-10-02/part-0.csv {"Day":"2026-10-02T00:00:00.0000000Z"}| Uri | Partition |
|---|---|
| /kustodata/lake/orders/day=2026-10-01/part-0.csv | {"Day": "2026-10-01T00:00:00.0000000Z"} |
| /kustodata/lake/orders/day=2026-10-02/part-0.csv | {"Day": "2026-10-02T00:00:00.0000000Z"} |
Two files, day=2026-10-01/part-0.csv and day=2026-10-02/part-0.csv, each mapped to its Day.
In Azure, the path is a storage URL, and the cluster's managed identity reads it, once its policy allows external tables:
// in Azure: a storage URL, read with the cluster managed identity
// (its managed_identity policy must allow the ExternalTable usage)
.create external table OrdersLake (OrderId:long, Customer:string,
Amount:real)
kind=storage partition by (Day:datetime)
pathformat=('day=' datetime_pattern('yyyy-MM-dd', Day))
dataformat=parquet (
h@'https://stshop.blob.core.windows.net/lake;managed_identity=system'
)When to use it:
- Good for: archives, occasional joins, backfills with .set-or-append
- Slower than ingested data: and results aren't cached
- Hot dashboards: ingest the data instead
- Delta tables: a query acceleration policy can help
Question 4: Someone asks how Data Factory, Power BI and Grafana connect to Azure Data Explorer. What do you use?
All three have built-in connectors that send KQL to the cluster. Each one signs in with its own identity.
| Tool | How it connects | Identity |
|---|---|---|
| Data Factory | linked service: Copy source and sink, Lookup | service principal or managed identity |
| Power BI | Azure Data Explorer connector: Import or DirectQuery | Microsoft Entra ID sign-in |
| Grafana | Azure Data Explorer Datasource plugin | Managed Grafana identity: database Viewer |
| Spark, Databricks | open-source Spark connector | Entra ID app (service principal) |
Power BI uses Import for smaller tables and DirectQuery for large, fresh fact tables. Grafana's plugin is preinstalled in Azure Managed Grafana.
My habit: every tool calls a stored function, not raw tables. Let's create RevenuePerHour, which takes a time range, in a Dashboards folder with a docstring. Power BI and Grafana pass their own range, and the logic lives in one place, under version control. We call it for 08:00 to 11:59 on October 20.
.create-or-alter function with (folder='Dashboards',
docstring='Orders and revenue per hour')
RevenuePerHour(fromTime:datetime, toTime:datetime) {
Sales
| where OrderTime between (fromTime .. toTime)
| summarize Orders = count(), Revenue = sum(Amount)
by Hour = bin(OrderTime, 1h)
}
RevenuePerHour(datetime(2026-10-20 08:00), datetime(2026-10-20 11:59))
| order by Hour asc$ kql -q -f dash.kql
> .create-or-alter function with (folder='Dashboards',
ok: .create-or-alter function with (folder='Dashboards',
> RevenuePerHour(datetime(2026-10-20 08:00), datetime(2026-10-20 11:59))
Hour Orders Revenue
-------------------- ------ -------
2026-10-20T08:00:00Z 2 39.5
2026-10-20T09:00:00Z 1 40
2026-10-20T10:00:00Z 2 69.5
2026-10-20T11:00:00Z 1 33| Hour | Orders | Revenue |
|---|---|---|
| 2026-10-20T08:00:00Z | 2 | 39.5 |
| 2026-10-20T09:00:00Z | 1 | 40 |
| 2026-10-20T10:00:00Z | 2 | 69.5 |
| 2026-10-20T11:00:00Z | 1 | 33 |
Four hours: 2 orders and 39.5 at 08:00, 1 and 40 at 09:00, 2 and 69.5 at 10:00, and 1 and 33 at 11:00.
A Data Factory copy from Azure Data Explorer truncates at 500,000 records or 64 MB unless the query sets notruncation, and its query timeout defaults to 10 minutes.
Question 5: A teammate needs query results exported to storage on a schedule. How does continuous export work?
Continuous export runs a query on an interval and writes the results to an external table, as CSV, TSV, JSON or Parquet. Database cursors make each run export only the rows that arrived since the last one.
Here is the flow:
- Sales: new rows since the cursor
- Continuous export: every 10 minutes
- External table: Parquet files in storage
The cursor gives exactly once for the files it reports. That's why the query has no time filter: the cursor does that job. The source needs the ingestion time policy, which new tables have by default, and you need Database Admin to create the export.
Let's define it on the emulator: a Parquet external table, SalesExport, and an export of Sales every 10 minutes. Then .show continuous-exports lists it.
.create external table SalesExport (OrderId:long, Customer:string,
Amount:real, OrderTime:datetime)
kind=storage dataformat=parquet
( h@'/kustodata/export/sales' )
.create-or-alter continuous-export SalesToLake
over (Sales) to table SalesExport
with (intervalBetweenRuns=10m)
<| Sales$ kql -q -f export.kql
> .create external table SalesExport (OrderId:long, Customer:string,
ok: .create external table SalesExport (OrderId:long, Customer:string,
> .create-or-alter continuous-export SalesToLake
ok: .create-or-alter continuous-export SalesToLake.show continuous-exports | project Name, ExternalTableName, Query, IntervalBetweenRuns, IsDisabled| Name | ExternalTableName | Query | IntervalBetweenRuns | IsDisabled |
|---|---|---|---|---|
| SalesToLake | SalesExport | Sales | 00:10:00 | 0 |
SalesToLake writes to SalesExport, runs the query Sales every 00:10:00, and IsDisabled is 0, which means false. The export runs themselves don't work on the emulator, so we stop at the definition.
In Azure, the external table points to storage and the export runs with the cluster's managed identity:
// in Azure: SalesExport's storage URL ends in ;impersonate
.alter-merge database shop policy managed_identity
```[{"ObjectId": "system", "AllowedUsages": "AutomatedFlows"}]```
.create-or-alter continuous-export SalesToLake
over (Sales) to table SalesExport
with (intervalBetweenRuns=10m, managedIdentity="system")
<| SalesThree things to plan for:
- Duplicates possible: after a failure mid-export
- Disabled after 7 days: of permanent failures
- Alert: continuous export metrics
Question 6: Someone asks how to deploy tables, functions and policies safely across environments. What do you use?
Treat the schema as code: idempotent scripts in git, deployed by a pipeline. Run the same script twice and nothing breaks.
First, .show database shop schema as csl script prints every table with its policies, and every function, external table and continuous export, as commands. So you keep them in git and compare environments.
$ kql --cell 98 '.show database shop schema as csl script'
DatabaseSchemaScript
--------------------------------------------------------------------------------------------------
.create-merge table Sales (OrderId:long, Customer:string, Amount:real, OrderTime:datetime)
.create-merge table Telemetry (DeviceId:string, Temp:real, Timestamp:datetime)
.create-merge table Orders (OrderId:long, Customer:string, Amount:real)
.create-or-alter function with (folder = "Dashboards", docstring = "Orders and revenue per hour...
.create-or-alter external table OrdersLake (OrderId:long, Customer:string, Amount:real) kind =...
.create-or-alter external table SalesExport (OrderId:long, Customer:string, Amount:real, OrderT...
.create-or-alter continuous-export SalesToLake over (Sales) to table SalesExport with (forcedLa...The tables come back as .create-merge table for Sales, Telemetry and Orders, then .create-or-alter for the function, the two external tables OrdersLake and SalesExport, and the continuous export SalesToLake. .create-merge adds what's missing and never drops a column. .create-or-alter replaces a function or an export.
Then, .execute database script runs commands in order and stops at the first failure. Our small runner, deploy.py, sends three: create a Returns table, drop a table that doesn't exist, and create a function.
.execute database script <|
.create-merge table Returns (OrderId:long, Reason:string)
.drop table OrdersStaging
.create-or-alter function ReturnsByReason() { Returns | count }import sys
from azure.kusto.data import KustoClient, KustoConnectionStringBuilder
# a tiny deploy runner; emulator only: no sign-in
client = KustoClient(KustoConnectionStringBuilder.with_no_authentication(
"http://localhost:8080"))
script = open(sys.argv[1]).read()
for row in client.execute_mgmt("shop", script).primary_results[0]:
print(f'{row["CommandType"]:<22} {row["Result"]:<10} {row["Reason"]}')
names = [r["Name"] for r in
client.execute_mgmt("shop", ".show functions").primary_results[0]]
print("ReturnsByReason deployed:", "ReturnsByReason" in names)$ python deploy.py deploy.kql
TableCreate Completed
TableDrop Failed Entity ID 'OrdersStaging' of kind 'Table' was not found.
ReturnsByReason deployed: FalseTableCreate is Completed. TableDrop Failed with Entity ID 'OrdersStaging' of kind 'Table' was not found. The function after it never ran: ReturnsByReason deployed: False. Nothing was rolled back, so the Returns table stays. Scripts must be safe to rerun.
In Azure, I deploy with a Bicep database script resource:
resource schema 'Microsoft.Kusto/clusters/databases/scripts@2023-05-02' = {
parent: shopDb
name: 'shop-schema'
properties: {
scriptContent: loadTextContent('shop_schema.kql')
continueOnErrors: false // the default
forceUpdateTag: buildId // a new value runs it again
}
}The options:
- Bicep database scripts: .create, .create-or-alter, .create-merge, .alter, .alter-merge, .add only
- Azure DevOps: Azure Data Explorer pipeline tools tasks
- Delta Kusto: open-source community tool, not a supported Azure product
Scenario 7: Data stopped arriving in a table for two hours and nobody noticed. What do you put in place?
A freshness check per table that pages someone. The question isn't whether the cluster is up; it's whether each table gets new data within its normal delay.
Let's create Freshness. It finds the last event in each table and flags the ones older than the allowed lag. We pin now to noon on October 20 and allow 30 minutes.
.create-or-alter function with (folder='Monitoring')
Freshness(asOf:datetime, maxLag:timespan) {
union withsource=TableName Sales, Telemetry
| extend EventTime = coalesce(OrderTime, Timestamp)
| summarize LastEvent = max(EventTime) by TableName
| extend Lag = asOf - LastEvent
| extend Stale = Lag > maxLag
}
Freshness(datetime(2026-10-20 12:00), 30m)
| order by TableName asc$ kql -q -f freshness.kql
> .create-or-alter function with (folder='Monitoring')
ok: .create-or-alter function with (folder='Monitoring')
> Freshness(datetime(2026-10-20 12:00), 30m)
TableName LastEvent Lag Stale
--------- -------------------- -------- -----
Sales 2026-10-20T11:50:00Z 00:10:00 false
Telemetry 2026-10-20T09:55:00Z 02:05:00 true| TableName | LastEvent | Lag | Stale |
|---|---|---|---|
| Sales | 2026-10-20T11:50:00Z | 00:10:00 | false |
| Telemetry | 2026-10-20T09:55:00Z | 02:05:00 | true |
Sales last saw an order at 11:50, a lag of 00:10:00, so Stale is false. Telemetry stopped at 09:55, a lag of 02:05:00: two hours five minutes, Stale is true.
In production, I use now() and the ingestion time, with the ingestion time policy, on by default, and the allowed lag per table in a config table. A scheduled alert, like an Azure Monitor log alert or a Logic App, runs it every few minutes:
// production: ingestion time and a lag per table from a config table
union withsource=TableName Sales, Telemetry
| summarize LastIngest = max(ingestion_time()) by TableName
| lookup kind=inner (FreshnessConfig) on TableName
| extend Stale = now() - LastIngest > MaxLag
| where StaleAround it, metric alerts on ingestion result failures, discovery latency, and events processed dropping to zero, plus the failed ingestion logs:
# the event hub data connection processes nothing for 30 minutes
az monitor metrics alert create -n adx-no-events -g rg-data \
--scopes "$ADX_ID" --condition "total EventsProcessed < 1" \
--window-size 30m --evaluation-frequency 5m --action "$AG_ID"The full checklist:
- Freshness check per table: a scheduled alert that pages someone
- Metric alerts: ingestion result, discovery latency, events processed = 0
- Compare: event hub incoming vs events processed
- Runbook: who is paged, where to look, how to backfill
- Test it: pause a feed and wait for the page
Key points to remember
- Metrics + alerts: keep alive, cache utilization factor, ingestion latency and result, throttling
- .show ingestion failures: management-command loads, 14 days; metrics and logs for the rest
- External tables query the lake in place; ingest what dashboards need
- Tools call stored functions; continuous export uses database cursors, no time filter
- Schema as idempotent scripts; a freshness check per table catches silence
The full script
Here is a Python script that writes every file and runs every query in order on the Kusto emulator; start the emulator container first, as the comments at the top of the script show.
Show the full script that writes the files and runs every step
# ========================================================================
# Azure Data Explorer Interview Q&A #9 - Operations, monitoring and integration
# TechBrothersIT. Names and sample data are invented for teaching.
# Every step below was run on the Kusto emulator (Kustainer, the Azure Data Explorer engine in a Linux container,
# build 2026.11.02); the outputs in the video and the blog post are the real outputs.
# To run it yourself (Docker on Linux, macOS or Windows with WSL), in an EMPTY folder:
# 1. Start the emulator with this folder mounted at /kustodata (the steps ingest files from there):
# docker run -d --name kustainer -e ACCEPT_EULA=Y -m 4G -p 8080:8080 -v "$PWD":/kustodata \
# mcr.microsoft.com/azuredataexplorer/kustainer-linux:latest
# Wait until `docker logs kustainer` says the database "is answering queries".
# 2. python KQL_Interview_QA_09_Code.py (Python 3.10 or later; kql.py itself needs no packages: it uses the REST API)
# The python scripts in some steps use the Kusto Python SDK: pip install azure-kusto-data azure-kusto-ingest
# "kql" in the commands is a small helper written by Step 0 (kql.py): it sends a query or a management command
# to http://localhost:8080 and prints the result as a text table (no sign-in: the emulator has no authentication).
# The emulator is for development and testing only; an Azure Data Explorer cluster needs Microsoft Entra ID sign-in
# (use the Kusto SDK or the web UI there).
# Steps that restart the emulator say so in a comment: run "docker restart kustainer" at that point.
# ========================================================================
import os, subprocess, sys, time
def write(path, text):
if text is None:
if os.path.exists(path): os.remove(path)
return
os.makedirs(os.path.dirname(path) or ".", exist_ok=True)
with open(path, "w", encoding="utf-8") as f:
f.write(text.strip("\n") + "\n")
def run(cmd, expect_error=False, allow_error=False, env=None):
print("\n$ " + cmd, flush=True)
rc = subprocess.run(cmd, shell=True, env=dict(os.environ, **(env or {}))).returncode
if rc != 0 and not (expect_error or allow_error):
raise SystemExit("stopped: this command was expected to succeed")
# Environment used by every step (the database the steps use)
os.environ.update({'KQL_DB': 'shop'})
# Step 0: the kql helper, the database and the starting files
write("kql.py", '"""kql: run KQL queries and management commands on the local Kusto emulator (Kustainer), print the results as text tables.\n kql \'StormEvents | take 3\' run one query or one management command (starts with \'.\')\n kql -f script.kql run every statement in the file (statements are separated by blank lines;\n // comment lines are skipped); each result is printed under its first line\n options: -d DB (default $KQL_DB or NetDefaultDB) --max N rows to print (default 30) --cell N max chars per cell (default 60)\n -q for management commands: print one line ("ok: <command>") instead of the result table\n --stats after a query, print the rows / extents the query scanned (from the engine\'s own query statistics)\n --props \'{"Options": {...}}\' client request properties (e.g. truncationmaxrecords)\n -e keep going after an error (the error is printed and the exit code is 1 at the end)\n --json print the last statement\'s first result as JSON {"cols": [...], "rows": [...]} (used by the video generator)\nEmpty cells: a null value in a non-string column prints as "null"; an empty or null string prints as nothing.\nA query that fails after returning some rows (for example a result over the truncation limit) prints the rows it got, then\n"Error (partial result): <engine message>", and exits 1.\nThe engine answers at http://localhost:8080 (no authentication: the emulator has none)."""\nimport json, os, re, sys, urllib.request, urllib.error\n\nURL = os.environ.get(\'KQL_URL\', \'http://localhost:8080\')\n\n\ndef post(path, db, csl, props=None):\n body = {\'db\': db, \'csl\': csl, \'properties\': json.dumps(props or {\'Options\': {}})}\n req = urllib.request.Request(URL + path, data=json.dumps(body).encode(), method=\'POST\',\n headers={\'Content-Type\': \'application/json; charset=utf-8\', \'Accept\': \'application/json\',\n \'x-ms-app\': \'kql-cli\', \'x-ms-client-request-id\': \'kql-cli\'})\n opener = urllib.request.build_opener(urllib.request.ProxyHandler({}))\n try:\n with opener.open(req, timeout=600) as r: return json.loads(r.read().decode(\'utf-8\'))\n except urllib.error.HTTPError as e:\n txt = e.read().decode(\'utf-8\', \'replace\')\n try:\n er = json.loads(txt)[\'error\']\n msg = er.get(\'@message\') or er.get(\'message\')\n except Exception:\n msg = txt.strip().split(\'\\n\')[0]\n msg = re.sub(r\'^Request is invalid and cannot be processed: \', \'\', msg)\n msg = re.sub(r\'\\s*\\(Activity stack:.*\', \'\', msg, flags=re.S)\n raise KqlError(msg.strip())\n\n\nclass KqlError(Exception): pass\n\n\ndef cell(v, width, typ=\'\'):\n if v is None: s = \'\' if typ in (\'string\', \'\') else \'null\'\n elif isinstance(v, bool): s = \'true\' if v else \'false\'\n elif isinstance(v, (dict, list)): s = json.dumps(v, separators=(\',\', \':\'), ensure_ascii=False)\n elif isinstance(v, float):\n s = repr(v)\n if s.endswith(\'.0\'): s = s[:-2]\n else: s = str(v)\n s = s.replace(\'\\r\', \' \').replace(\'\\n\', \' \')\n if len(s) > width: s = s[:width - 3] + \'...\'\n return s\n\n\ndef table(cols, rows, mx, width, types=None):\n rows = rows if isinstance(rows, list) else []\n types = types or [\'\'] * len(cols)\n shown = [[cell(v, width, types[i]) for i, v in enumerate(r)] for r in rows[:mx]]\n w = [max([len(c)] + [len(r[i]) for r in shown]) for i, c in enumerate(cols)]\n out = [\' \'.join(c.ljust(w[i]) for i, c in enumerate(cols)).rstrip(),\n \' \'.join(\'-\' * w[i] for i in range(len(cols)))]\n for r in shown:\n out.append(\' \'.join(v.ljust(w[i]) for i, v in enumerate(r)).rstrip())\n n = len(rows)\n if n > mx: out.append(\'... %d more rows (%d rows in total)\' % (n - mx, n))\n elif n == 0: out.append(\'(no rows)\')\n return \'\\n\'.join(out)\n\n\nLAST = {}\n\n\ndef one_api_msg(errs):\n msgs = []\n for e in errs or []:\n er = e.get(\'error\', e) if isinstance(e, dict) else {}\n m = er.get(\'@message\') or er.get(\'message\') or str(e)\n m = re.sub(r\'\\s*\\(Activity stack:.*\', \'\', m, flags=re.S).strip()\n if m not in msgs: msgs.append(m)\n return \' | \'.join(msgs) or \'unknown error\'\n\n\ndef run_query(db, csl, a):\n frames = post(\'/v2/rest/query\', db, csl, a.props)\n outs, stats, partial = [], None, None\n for f in frames:\n if f.get(\'FrameType\') == \'DataSetCompletion\' and f.get(\'HasErrors\') and partial is None:\n partial = one_api_msg(f.get(\'OneApiErrors\'))\n if f.get(\'FrameType\') != \'DataTable\': continue\n if f.get(\'TableKind\') == \'PrimaryResult\':\n rows = [r for r in f[\'Rows\'] if isinstance(r, list)]\n for r in f[\'Rows\']:\n if isinstance(r, dict) and r.get(\'OneApiErrors\'): partial = one_api_msg(r[\'OneApiErrors\'])\n cols = [c[\'ColumnName\'] for c in f[\'Columns\']]\n if \'cols\' not in LAST: LAST.update(cols=cols, rows=rows)\n outs.append(table(cols, rows, a.max, a.cell, [str(c.get(\'ColumnType\', \'\')).lower() for c in f[\'Columns\']]))\n if f.get(\'TableKind\') == \'QueryCompletionInformation\':\n cols = [c[\'ColumnName\'] for c in f[\'Columns\']]\n for r in f[\'Rows\']:\n d = dict(zip(cols, r))\n if d.get(\'EventTypeName\') == \'QueryResourceConsumption\':\n try: stats = json.loads(d[\'Payload\'])\n except Exception: pass\n if d.get(\'Level\') in (2, 3) and d.get(\'EventTypeName\') not in (\'QueryResourceConsumption\',):\n txt = d.get(\'Payload\') or \'\'\n try: txt = json.loads(txt).get(\'Text\', txt)\n except Exception: pass\n outs.append(\'%s: %s\' % (\'Warning\' if d.get(\'Level\') == 3 else \'Error\', txt))\n if a.stats and stats:\n sh = (stats.get(\'resource_usage\') or {}).get(\'cache\', {}).get(\'shards\', {})\n ix = stats.get(\'input_dataset_statistics\') or {}\n ex, rw = ix.get(\'extents\') or {}, ix.get(\'rows\') or {}\n outs.append(\'scanned: %s of %s extents, %s of %s rows\' % (ex.get(\'scanned\'), ex.get(\'total\'), rw.get(\'scanned\'), rw.get(\'total\')))\n if partial:\n outs.append(\'Error (partial result): \' + partial)\n raise PartialError(\'\\n\\n\'.join(outs))\n return \'\\n\\n\'.join(outs)\n\n\nclass PartialError(Exception): pass\n\n\ndef run_mgmt(db, csl, a):\n d = post(\'/v1/rest/mgmt\', db, csl, a.props)\n if a.q and not a.json: return \'ok: \' + first_line(csl)\n t = (d.get(\'Tables\') or [{}])[0]\n LAST.update(cols=[c[\'ColumnName\'] for c in t.get(\'Columns\', [])], rows=t.get(\'Rows\', []))\n return table([c[\'ColumnName\'] for c in t.get(\'Columns\', [])], t.get(\'Rows\', []), a.max, a.cell,\n [str(c.get(\'DataType\', c.get(\'ColumnType\', \'\'))).lower().replace(\'system.\', \'\') for c in t.get(\'Columns\', [])])\n\n\ndef first_line(s):\n l = s.strip().split(\'\\n\')[0].strip()\n return l if len(l) <= 90 else l[:87] + \'...\'\n\n\ndef statements(text):\n blocks, cur = [], []\n for l in text.split(\'\\n\'):\n if not l.strip():\n if cur: blocks.append(\'\\n\'.join(cur)); cur = []\n continue\n if l.strip().startswith(\'//\') and not cur: continue\n cur.append(l)\n if cur: blocks.append(\'\\n\'.join(cur))\n return blocks\n\n\ndef main():\n import argparse\n p = argparse.ArgumentParser(add_help=False)\n p.add_argument(\'-d\', default=os.environ.get(\'KQL_DB\', \'NetDefaultDB\'))\n p.add_argument(\'-f\'); p.add_argument(\'-q\', action=\'store_true\'); p.add_argument(\'-e\', action=\'store_true\')\n p.add_argument(\'--stats\', action=\'store_true\'); p.add_argument(\'--max\', type=int, default=30)\n p.add_argument(\'--cell\', type=int, default=60); p.add_argument(\'--props\'); p.add_argument(\'-h\', \'--help\', action=\'store_true\')\n p.add_argument(\'--json\', action=\'store_true\')\n p.add_argument(\'text\', nargs=\'?\')\n a = p.parse_args()\n if a.help or (not a.f and not a.text): print(__doc__); return 0\n a.props = json.loads(a.props) if a.props else None\n stmts = statements(open(a.f).read()) if a.f else [a.text]\n rc, first = 0, True\n if a.json:\n for s in stmts:\n LAST.clear()\n try:\n run_mgmt(a.d, s, a) if s.lstrip().startswith(\'.\') else run_query(a.d, s, a)\n except (KqlError, PartialError) as e:\n print(\'Error: \' + str(e)); return 1\n print(json.dumps(dict(cols=LAST.get(\'cols\', []), rows=LAST.get(\'rows\', [])), ensure_ascii=False))\n return 0\n for s in stmts:\n if a.f:\n if not first: print()\n print(\'> \' + first_line(s))\n first = False\n try:\n out = run_mgmt(a.d, s, a) if s.lstrip().startswith(\'.\') else run_query(a.d, s, a)\n if out: print(out)\n except PartialError as e:\n print(str(e)); rc = 1\n if not a.e: return 1\n except KqlError as e:\n print(\'Error: \' + str(e)); rc = 1\n if not a.e: return 1\n return rc\n\n\nif __name__ == \'__main__\':\n sys.exit(main())\n')
KQL = '"%s" "%s"' % (sys.executable, os.path.abspath("kql.py")) # commands below run it as: python kql.py ...
run(KQL + ' -d NetDefaultDB -q \'.create database shop persist (@"/kustodata/dbs/shop/md", @"/kustodata/dbs/shop/data")\'')
write('setup.kql', """\
.create table Sales (OrderId:long, Customer:string, Amount:real, OrderTime:datetime)
.ingest inline into table Sales <|
5001,C001,25.5,2026-10-20T08:05:00Z
5002,C002,14,2026-10-20T08:40:00Z
5003,C001,40,2026-10-20T09:15:00Z
5004,C003,9.5,2026-10-20T10:20:00Z
5005,C002,60,2026-10-20T10:55:00Z
5006,C004,33,2026-10-20T11:50:00Z
.create table Telemetry (DeviceId:string, Temp:real, Timestamp:datetime)
.ingest inline into table Telemetry <|
d1,21.5,2026-10-20T09:00:00Z
d2,22,2026-10-20T09:30:00Z
d1,21.8,2026-10-20T09:55:00Z
.create table Orders (OrderId:long, Customer:string, Amount:real)
""")
run(KQL + ' -q -f setup.kql')
# Step 1: Three bad loads: a short row, CSV sent as Parquet, a missing file; no error, HasErrors true, table empty
write('ingest.kql', """\
.ingest into table Orders (@"/kustodata/orders_1001.csv")
with (format='csv', ignoreFirstRecord=true)
| project ItemLoaded, HasErrors
.ingest into table Orders (@"/kustodata/orders_1002.csv")
with (format='parquet')
| project ItemLoaded, HasErrors
.ingest into table Orders (@"/kustodata/orders_1003.csv")
with (format='csv')
| project ItemLoaded, HasErrors
Orders | count
""")
write('orders_1001.csv', """\
OrderId,Customer,Amount
6001,C004,12
6002,C005
""")
write('orders_1002.csv', """\
6003,C006,20
6004,C001,18
""")
run(KQL + ' -f ingest.kql')
# Step 2: (wait until the engine has flushed its ingestion failures)
run('for i in $(seq 1 60); do n=$(kql \'.show ingestion failures | where Table == "Orders" | count\' | tail -1); [ "$n" -ge 3 ] && break; sleep 5; done; echo ingestion failures flushed')
# Step 3: .show ingestion failures for the Orders table
run(KQL + ' \'.show ingestion failures | where Table == "Orders" | project Source = IngestionSourcePath, FailureKind, ErrorCode | order by Source asc\'')
# Step 4: A partitioned external table over day folders; a filter on Day reads one folder
write('lake.kql', """\
.create external table OrdersLake (OrderId:long, Customer:string,
Amount:real)
kind=storage partition by (Day:datetime)
pathformat=('day=' datetime_pattern('yyyy-MM-dd', Day))
dataformat=csv
( h@'/kustodata/lake/orders' )
external_table('OrdersLake')
| where Day == datetime(2026-10-02)
| order by OrderId asc
""")
write('lake/orders/day=2026-10-01/part-0.csv', """\
5001,C001,25.5
5002,C002,14
""")
write('lake/orders/day=2026-10-02/part-0.csv', """\
5003,C001,40
5004,C003,9.5
5005,C002,60
""")
run(KQL + ' -q -f lake.kql')
# Step 5: The files behind the external table and their partitions
run(KQL + " '.show external table OrdersLake artifacts | project Uri, Partition | order by Uri asc'")
# Step 6: A stored function as the contract for Power BI and Grafana
write('dash.kql', """\
.create-or-alter function with (folder='Dashboards',
docstring='Orders and revenue per hour')
RevenuePerHour(fromTime:datetime, toTime:datetime) {
Sales
| where OrderTime between (fromTime .. toTime)
| summarize Orders = count(), Revenue = sum(Amount)
by Hour = bin(OrderTime, 1h)
}
RevenuePerHour(datetime(2026-10-20 08:00), datetime(2026-10-20 11:59))
| order by Hour asc
""")
run(KQL + ' -q -f dash.kql')
# Step 7: Continuous export: a Parquet external table and an export of Sales every 10 minutes (definition only)
write('export.kql', """\
.create external table SalesExport (OrderId:long, Customer:string,
Amount:real, OrderTime:datetime)
kind=storage dataformat=parquet
( h@'/kustodata/export/sales' )
.create-or-alter continuous-export SalesToLake
over (Sales) to table SalesExport
with (intervalBetweenRuns=10m)
<| Sales
""")
run(KQL + ' -q -f export.kql')
# Step 8: The whole database schema as a script of idempotent commands
run(KQL + " --cell 98 '.show database shop schema as csl script'")
# Step 9: .execute database script stops at the first failure; earlier commands stay, later ones never run
write('deploy.kql', """\
.execute database script <|
.create-merge table Returns (OrderId:long, Reason:string)
.drop table OrdersStaging
.create-or-alter function ReturnsByReason() { Returns | count }
""")
write('deploy.py', """\
import sys
from azure.kusto.data import KustoClient, KustoConnectionStringBuilder
# a tiny deploy runner; emulator only: no sign-in
client = KustoClient(KustoConnectionStringBuilder.with_no_authentication(
"http://localhost:8080"))
script = open(sys.argv[1]).read()
for row in client.execute_mgmt("shop", script).primary_results[0]:
print(f'{row["CommandType"]:<22} {row["Result"]:<10} {row["Reason"]}')
names = [r["Name"] for r in
client.execute_mgmt("shop", ".show functions").primary_results[0]]
print("ReturnsByReason deployed:", "ReturnsByReason" in names)
""")
run('python deploy.py deploy.kql')
# Step 10: A freshness function: last event per table vs an allowed lag, with now pinned to 2026-10-20 12:00
write('freshness.kql', """\
.create-or-alter function with (folder='Monitoring')
Freshness(asOf:datetime, maxLag:timespan) {
union withsource=TableName Sales, Telemetry
| extend EventTime = coalesce(OrderTime, Timestamp)
| summarize LastEvent = max(EventTime) by TableName
| extend Lag = asOf - LastEvent
| extend Stale = Lag > maxLag
}
Freshness(datetime(2026-10-20 12:00), 30m)
| order by TableName asc
""")
run(KQL + ' -q -f freshness.kql')
Part 10: Real-World Manager Scenarios
The sample data on this page is invented for teaching. Every terminal output is the real output of the command above it, run on the Kusto emulator (Kustainer, the Azure Data Explorer engine in a Linux container, build 2026.11.02). Features the emulator cannot show truthfully are shown as code. Part of the Real Time Azure Data Explorer and KQL Interview Questions and Answers series on TechBrothersIT. Features, limits and their status (GA or preview) are current as of November 2026; check the official Azure Data Explorer and Kusto documentation (learn.microsoft.com/azure/data-explorer and learn.microsoft.com/kusto) for anything you quote in an interview.
No comments:
Post a Comment
Note: Only a member of this blog may post a comment.