Introduction
There are many use cases for retrieving data from an API within Fabric. This post describes the requirements and PySpark code to query the Microsoft Graph API to retrieve Azure resource data, than store the results in a Lakehouse table with a partition based on the year.
A practical use case for this activity would be to store Azure resource details beyond the available 30 days provided by the Microsoft Graph. This would facilitate long-term analysis of resource usage and configuration change to identify resource-creep or configuration drift.
One approach to host this function would be to use Azure Fabric and leverage data pipelines for orchestration and Notebooks for function logic. Azure Fabric is an enterprise-grade data platform SaaS built on OneLake, Fabric’s centralised organisation-level storage, akin to OneDrive.
For brevity, this post will focus on the PySpark Notebook code snippets that will retrieve the Azure resource data from Microsoft Graph. However, for context, a Fabric workspace would be setup with a connection to an existing Lakehouse where the resource tables would be hosted. The tables are native Spark tables that are highly efficient for both cost and performance so are ideal for the size of data that would be accumulated over time.
Requirements
This function will require an Entra ID Registered Application (App) that has ‘Read’ RBAC access to the root Management Group (tenant-wide) or subscription(s) that resources details are to be collected.
The App will use the ‘client credential’ OAuth flow to retrieve an access token from Microsoft that will authorise the reading of resource data from the Graph.
High-Level Steps
Step 1 – Import Python libraries required for the function:
# import Python libraries
import requests
import json
from datetime import date
from pyspark.sql.functions import lit
Step 2 – Get an OAuth token from Microsoft to authorise access to the Graph resources. The authorisation will be based on the App’s RBAC ‘Read’ permissions.
The ‘client_id’, ‘client_secret’, and ‘tenant_id’ can be passed through to the Notebook using the pipeline parameters.
These parameters may be set in the ‘Parameters’ section of the pipeline, with the ‘client_secret’ stored as a secure value. Alternatively, these values can be stored in a Key Vault and retrieved using the pipeline ‘Web’ activity and a workspace ‘Managed Identity’. The workspace managed identity is setup under the workspace settings and must be granted the ‘Key Vault Secrets User’ RBAC role on the Key Vault.
If using the pipeline parameter method, the pipeline parameters are mapped through to the Notebook via ‘Settings > Base Parameters’ using dynamic content expressions, e.g.,
Name: app_id
Type: String
Value: @pipeline().parameters.app_id
# Get token for Graph API
print("Get OAuth token for application...")
token_url = f"https://login.microsoftonline.com/{tenant_id}/oauth2/token"
token_data = {
"client_id": app_id,
"client_secret": app_secret,
"resource": "https://management.azure.com/",
"grant_type": "client_credentials"
}
token_headers = {
"Content-Type": "application/x-www-form-urlencoded"
}
token_response = requests.post(token_url, data=token_data, headers=token_headers)
access_token = token_response.json().get("access_token")
if access_token is None:
print(" OAuth Token - Fail")
else:
print(" OAuth Token - Success")
Step 3 – Read the resource types to be inventoried from a text file stored in the Lakehouse.
The ‘resource type’ is used in KQL query sent to the Graph API to retrieve resources, e.g.,
query Resources | where type == ‘Microsoft.Compute/virtualMachines’
There a number of ways to store the resource types; csv file, table, or alternatively generate the list dynamically as part of the function. The latter approach ensures all resource types currently in use within the tenant are retrieved, although there may be a use case to only collect data for selected resource types. One way to dynamically create the ‘resource types’ list, is to query the Graph API first with the KQL query: query = “Resources | distinct type” and store the values in an array.
The following snippet reads a file stored in the Lakehouse into an array:
# Define the relative path
file_path = f"Files/Resource_Types_to_be_Inventoried.txt"
print(f" Relative path: {file_path}")
# Read the text file into a DataFrame
df = spark.read.text(file_path)
# Show the contents
df.show(truncate=False)
# Extract the 'value' column as a list
values = df.select("value").rdd.map(lambda row: row.value).collect()
Step 4 – Loop through the array of ‘resource types’ to be inventoried.
The ‘for each’ loop calls two functions that must be defined in the Notebook before this code snippet :
1. retrieve_resources(resource_type, access_token):
2. write_resources_to_table(resources, table_name, year):
# Iterate over each resource to be inventoried
print("Query Graph API for resource data...")
today = date.today()
year = today.year # used for table partition
for item in values:
print(f" Processing: {item}")
resource_type = item.lower()
resource_type_short = resource_type.split("/")[1]
resources = retrieve_resources(resource_type, access_token)
if resources:
print(f" {len(resources)} items found")
table_name = f"{resource_type_short}_{today}"
table_name = table_name.replace("-", "_")
result = write_resources_to_table(resources, table_name, year)
if result:
print(f" Resources successfully written to '{table_name}' partition 'year={year}'")
else:
print(f" Resources failed to be written to '{table_name}' partition 'year={year}'")
else:
print(f" Unable to retrieve resources for '{resource_type_short}'")
Function 1 – retrieve_resources
This function also checks for resource_types that don’t includes a ‘subscriptionId’ field and extracts the subscription ID from the ‘resource id’ attribute.
# FUNCTION: Query Azure Graph for the resource data
def retrieve_resources(resource_type, access_token):
table = "Resources"
resources_url = "https://management.azure.com/providers/Microsoft.ResourceGraph/Resources?api-version=2021-03-01"
headers = {
"Authorization": f"Bearer {access_token}",
"Content-Type": "application/json"
}
# array to store resource objects
resources = []
skip_token = None # pagination
try:
while True:
body = {
"query": f"{table} | where type == '{resource_type}'",
"options": {
"$skipToken": skip_token
}
}
response = requests.post(url=resources_url, headers=headers, json=body)
response.raise_for_status()
result = response.json()
# process returned resource objects
for data in result.get("data", []):
if "subscriptionId" not in data or not data["subscriptionId"]:
# extract subscriptionId from the resource Id
if "id" in data and data["id"]:
parts = data["id"].split("/")
subscription = parts[2] if len(parts) > 2 else None
else:
subscription = None
data["subscriptionId"] = subscription
resources.append(data)
skip_token = result.get("$skipToken")
if not skip_token:
break
return resources
except Exception as e:
print(f". Skipping - unable to process: {resource_type}")
print(f"Error: {e}")
return None
Function 2 – write_resources_to_table
After the ‘retrieve_resources’ function returns an array of resources, the second function is called to write the data to the Lakehouse table.
The table is partitioned by ‘year’ to improve query performance over years. Partitioning is optional, and this partition is included to demonstrate the method. An alternate partition could be by ‘resource type’ or nested ‘year > resource type’.
# FUNCTION: Write resource data to Lakehouse table
def write_resources_to_table(resources, table_name, year):
try:
json_strings = [json.dumps(obj, default=list) for obj in resources]
# Parallelize and read as DataFrame
rdd = spark.sparkContext.parallelize(json_strings)
df = spark.read.json(rdd)
df = convert_nested_to_json(df) # convert nested json (1 level)
df = df.withColumn("year", lit(year))
df.write.format("delta").mode("overwrite").partitionBy("year").saveAsTable(table_name)
return True
except Exception as e:
print(f"Error: {e}")
return False
Function 3 – convert_nested_to_json
This function takes the dataframe that contains the array of json-formatted resources and converts any first-level nested json objects e.g. the resource ‘properties’ into a column containing the json object i.e. flattens the json object down to the first-level.
# FUNCTION: convert all nested columns to JSON strings
def convert_nested_to_json(df):
"""
Converts all StructType, ArrayType, and MapType columns into JSON string columns.
"""
transformed_cols = []
for field in df.schema.fields:
if field.dataType.simpleString().startswith(("struct", "array", "map")):
# Convert nested column to JSON string
transformed_cols.append(to_json(col(field.name)).alias(field.name))
else:
# Keep primitive column as-is
transformed_cols.append(col(field.name))
return df.select(transformed_cols)
0 Comments