I have previously explained how to query a KQL database in a notebook using the Kusto Spark connector, Kusto Python SDK, and KQLMagic. Now, let's explore another method using the REST API. Although this is covered in the ADX documentation, it isn't in Fabric (with example), so I wanted to write a quick blog to show how you can query a table from an Eventhouse using a REST API.
Requirements:
import requests
import json
import pandas as pd
from datetime import datetime
def query_kusto(cluster_url: str, database: str, query: str, token: str):
"""
Function to query Fabric Kusto db using REST API
Args:
cluster_url: Full cluster URL (e.g., 'https://guid.kusto.fabric.microsoft.com')
database: Database name
query: KQL query
token: Bearer token
"""
headers = {
'Authorization': f'Bearer {token}',
'Content-Type': 'application/json',
'Accept': 'application/json'
}
body = {
'db': database,
'csl': query
}
response = requests.post(
f'{cluster_url}/v1/rest/query',
headers=headers,
json=body
)
return response.json()
Example
cluster_url = 'https://<guid>.z4.kusto.fabric.microsoft.com'
database = '<dbname>'
kqlquery = '''
table | take 2
'''
token = notebookutils.credentials.getToken(cluster_url)
try:
results = query_kusto(cluster_url, database, kqlquery, token)
except Exception as e:
print(f"Error querying Kusto: {str(e)}")
This will return the result as JSON text and we need to parse the output to create a dataframe for further analysis. It also returns the data types.
Parse the output:
Below we parse the JSON and assign datatypes (string, numeric, datetime only. For dynamic and other dtypes, assign dtypes as required).
def kusto_to_df(results):
"""
Convert Kusto query results to pandas DataFrame with appropriate data types.
Unrecognized types default to string.
Args:
results (dict): Kusto query results containing Tables, Columns, and Rows
Returns:
pandas.DataFrame: DataFrame with properly typed columns
"""
columns_info = results['Tables'][0]['Columns']
rows = results['Tables'][0]['Rows']
df = pd.DataFrame(rows, columns=[col['ColumnName'] for col in columns_info])
for col_info in columns_info:
col_name = col_info['ColumnName']
data_type = col_info['DataType']
try:
if data_type == 'DateTime':
df[col_name] = pd.to_datetime(df[col_name])
elif data_type in ['Int64', 'Int32', 'Long']:
df[col_name] = pd.to_numeric(df[col_name], errors='coerce').fillna(0).astype('int64')
elif data_type == 'Real' or data_type == 'Double':
df[col_name] = pd.to_numeric(df[col_name], errors='coerce')
else:
df[col_name] = df[col_name].astype(str)
except Exception as e:
print(f"Warning: Could not convert column {col_name} to {data_type}, defaulting to string: {str(e)}")
df[col_name] = df[col_name].astype(str)
return df
kusto_to_df(results)
Notes
Unlike Kusto spark connector, KQLMagic etc, this is language agnostic as long as you can generate a token and call an API. You can query in an external application as well.
This not meant to be used to return large amount of data. I am not sure of the limits, but I am sure just like other Fabric APIs, this endpoint also has rate and size limitations.
You can also ingest data using an API with /v1/rest/ingest URI.
REST API overview - Kusto | Microsoft Learn