Building True Tracing for Multi-Step AI Pipelines

Debugging traditional software applications relies on call stacks, HTTP status codes, and line by line log traces. Debugging non-deterministic LLM pipelines and multi-agent workflows is more complex. 

When an agent outputs an incorrect response or encounters an execution loop, traditional application logs often fail to explain why. True AI observability requires tracing prompt construction steps, model parameters, retrieved context chunks, tool call inputs/outputs, latency metrics, and exact token costs per step. 

In this post I will detail how to implement OpenTelemetry tracing using the Azure AI Inference SDK and Azure Application Insights within Microsoft Azure AI Foundry.

OpenTelemetry structures tracing by wrapping distinct operations in Spans. Spans form hierarchical trees that track execution flow across distributed microservices and LLM tool executions. 

Python Implementation: End-to-End Tracing 

Below is a complete implementation using OpenTelemetry SDK with Azure Monitor and Azure AI Project clients.

Python
import os
from opentelemetry import trace
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import BatchSpanProcessor
from azure.monitor.opentelemetry.exporter import AzureMonitorTraceExporter
from azure.ai.inference import ChatCompletionsClient
from azure.ai.inference.models import SystemMessage, UserMessage
from azure.core.credentials import AzureKeyCredential

# Initialize Telemetry Tracing
connection_string = os.getenv("APPLICATIONINSIGHTS_CONNECTION_STRING")
tracer_provider = TracerProvider()
exporter = AzureMonitorTraceExporter(connection_string=connection_string)
tracer_provider.add_span_processor(BatchSpanProcessor(exporter))
trace.set_tracer_provider(tracer_provider)

tracer = trace.get_tracer("foundry.ai.pipeline", "1.0.0")

def execute_traced_ai_workflow(user_query: str):
    with tracer.start_as_current_span("ai_workflow_execution") as parent_span:
        parent_span.set_attribute("ai.user_query", user_query)
        
        # Step 1: Trace Context Retrieval
        with tracer.start_as_current_span("rag_context_retrieval") as rag_span:
            retrieved_context = "System metric status: CPU load 88%."
            rag_span.set_attribute("ai.rag.retrieved_chunks_count", 1)
            rag_span.set_attribute("ai.rag.context_length", len(retrieved_context))
        
        # Step 2: Trace LLM Inference Call
        with tracer.start_as_current_span("azure_openai_inference") as llm_span:
            endpoint = os.getenv("AZURE_AI_INFERENCE_ENDPOINT")
            key = os.getenv("AZURE_AI_INFERENCE_CREDENTIAL")
            
            client = ChatCompletionsClient(endpoint=endpoint, credential=AzureKeyCredential(key))
            
            messages = [
                SystemMessage(content=f"Use context: {retrieved_context}"),
                UserMessage(content=user_query)
            ]
            
            llm_span.set_attribute("gen_ai.system", "azure_openai")
            llm_span.set_attribute("gen_ai.request.model", "gpt-4o")
            
            response = client.complete(messages=messages, model="gpt-4o")
            
            # Record Token Usage Attributes
            if hasattr(response, "usage"):
                llm_span.set_attribute("gen_ai.usage.prompt_tokens", response.usage.prompt_tokens)
                llm_span.set_attribute("gen_ai.usage.completion_tokens", response.usage.completion_tokens)
                llm_span.set_attribute("gen_ai.usage.total_tokens", response.usage.total_tokens)
                
            completion_text = response.choices[0].message.content
            llm_span.set_attribute("gen_ai.response.completion", completion_text)
            
            return completion_text

Key OpenTelemetry Semantic Conventions for AI

To ensure traces render cleanly in Azure AI Foundry dashboards, standard OpenTelemetry attribute naming conventions should be maintained:

Attribute Key	Description	Example
gen_ai.system	AI system or provider	azure_openai
gen_ai.request.model	Model deployment name requested	gpt-4o
gen_ai.usage.prompt_tokens	Input tokens consumed	452
gen_ai.usage.completion_tokens	Output tokens generated	128
ai.rag.retrieved_chunks_count	Number of context chunks retrieved	4

 Analyzing Traces for Root Cause Analysis

 With structured spans enabled:

 Identifying Tool Call Bottlenecks: Filter spans by duration to quickly identify API latency bottlenecks versus generation delays.

 Debugging Hallucinations: Inspect rag_context_retrieval spans alongside gen_ai.response.completion to verify whether incorrect outputs stemmed from missing source data or model hallucination.

Setting Up Automated Cost Guardrails for LLM Pipelines

Generative AI projects often start with predictable prototyping budgets. However, moving pipelines to production introduces unpredictable user input volumes, unconstrained multi turn conversations, agent loop cascades, and recursive retrieval steps. Without strict runtime cost controls, spending can escalate very quickly.

Managing costs effectively requires proactive, dynamic guardrails embedded into client SDK wrappers and gateway components. This blog post provides technical patterns using Python and TypeScript to implement token caps, cost estimation middleware, request throttling, and model fallback routing in Azure AI Foundry environments.

Anatomy of LLM Cost Drivers

Unexpected billing spikes generally trace back to three root causes:

Prompt Inflation: System instructions and full conversation histories attached to every turn without context truncation.

Agentic Loop Cascades: Autonomous agents hitting edge conditions and looping continuously against model endpoints.

Over-provisioned Model Selection: Routing basic text routing or classification tasks to large models like GPT-4o instead of smaller models like GPT-4o-mini.

Implementing Middleware Token Guardrails

Below is a Python implementation of an Azure AI Inference SDK client wrapped with dynamic token counting, prompt budget enforcement, and fallback model routing.

import sys
import tiktoken
from azure.ai.inference import ChatCompletionsClient
from azure.ai.inference.models import SystemMessage, UserMessage
from azure.core.credentials import AzureKeyCredential

class CostGuardedFoundryClient:
    def __init__(self, primary_endpoint: str, primary_key: str, fallback_endpoint: str, fallback_key: str):
        self.primary_client = ChatCompletionsClient(
            endpoint=primary_endpoint, credential=AzureKeyCredential(primary_key)
        )
        self.fallback_client = ChatCompletionsClient(
            endpoint=fallback_endpoint, credential=AzureKeyCredential(fallback_key)
        )
        self.tokenizer = tiktoken.get_encoding("cl100k_base")
        
        # Pricing constants per 1k tokens (example estimation rates)
        self.PRIMARY_INPUT_COST = 0.0025
        self.FALLBACK_INPUT_COST = 0.00015
        self.MAX_ALLOWED_PROMPT_TOKENS = 4000

    def _count_tokens(self, text: str) -> int:
        return len(self.tokenizer.encode(text))

    def generate_guarded_completion(self, system_prompt: str, user_prompt: str, max_cost_limit_usd: float):
        total_input_tokens = self._count_tokens(system_prompt) + self._count_tokens(user_prompt)
        
        # Guardrail 1: Enforce Token Truncation
        if total_input_tokens > self.MAX_ALLOWED_PROMPT_TOKENS:
            raise ValueError(f"Prompt token count ({total_input_tokens}) exceeds maximum limit ({self.MAX_ALLOWED_PROMPT_TOKENS}).")

        # Guardrail 2: Dynamic Routing Based on Budget & Token Count
        estimated_primary_cost = (total_input_tokens / 1000.0) * self.PRIMARY_INPUT_COST
        
        if estimated_primary_cost > max_cost_limit_usd:
            # Fallback to lower-cost model deployment
            print(f"Primary model estimated cost (${estimated_primary_cost:.4f}) exceeds target limit (${max_cost_limit_usd:.4f}). Routing to fallback model.")
            response = self.fallback_client.complete(
                messages=[
                    SystemMessage(content=system_prompt),
                    UserMessage(content=user_prompt)
                ],
                model="gpt-4o-mini"
            )
        else:
            response = self.primary_client.complete(
                messages=[
                    SystemMessage(content=system_prompt),
                    UserMessage(content=user_prompt)
                ],
                model="gpt-4o"
            )
            
        return response

TypeScript Implementation for Real Time Prompt Truncation

In Node.js applications interacting with Microsoft Azure AI Foundry, sliding-window message truncation prevents conversation histories from growing unbounded.

TypeScript
import { AzureKeyCredential } from "@azure/core-auth";
import ModelClient from "@azure-rest/ai-inference";

interface ChatMessage {
  role: "system" | "user" | "assistant";
  content: string;
}

export class TokenBudgetManager {
  private maxTokens: number;

  constructor(maxTokens: number = 3000) {
    this.maxTokens = maxTokens;
  }

  // Basic token estimation heuristic (approx 4 chars per token)
  private estimateTokens(text: string): int {
    return Math.ceil(text.length / 4);
  }

  public truncateHistory(systemPrompt: string, history: ChatMessage[]): ChatMessage[] {
    let currentTokens = this.estimateTokens(systemPrompt);
    const truncated: ChatMessage[] = [];

    for (let i = history.length - 1; i >= 0; i--) {
      const msgTokens = this.estimateTokens(history[i].content);
      if (currentTokens + msgTokens > this.maxTokens) {
        break;
      }
      currentTokens += msgTokens;
      truncated.unshift(history[i]);
    }

    return [{ role: "system", content: systemPrompt }, ...truncated];
  }
}

Architectural Cost Controls

Circuit Breakers: Implement rate limits using Azure API Management (APIM) placed in front of Azure OpenAI endpoints. Set quota keys by subscription ID or Entra ID user principal.

Semantic Caching: Store prompt-response pairs in Azure Cache for Redis to avoid executing duplicate queries against model endpoints.

Implementing Fine-Grained Row-Level Access Controls (RLAC) in Vector Search

Retrieval-Augmented Generation (RAG) unlocks powerful Enterprise search over unstructured document repositories. However, standard RAG implementations often face security challenges. In thousands of organisations, a security gap exists.  Once documents are ingested into a unified vector store, vector similarity search operates indiscriminately across security boundaries. An employee querying human resources policies might inadvertently retrieve vector chunks originating from executive compensation plans or restricted financial forecasts.

To make RAG enterprise-ready, identity aware document retrieval is mandatory. Security filters must execute directly within the retrieval query engine, ensuring users only retrieve chunks they are authorised to access under Active Directory / Entra ID policies.

This guide explores how to build fine-grained, row-level access control (RLAC) into Azure AI Search vector indexes connected to Microsoft Azure AI Foundry workloads using Python and TypeScript.

The Security Gap in Standard Vector Retrieval

In naive RAG deployments, document chunking and vector indexing strip away access control lists (ACLs). 

Consider an index containing the below for example:

• Public HR policy documents (Group: Everyone)
• Regional manager operational guidelines (Group: Regional_Managers)
• Sensitive salary metadata (Group: Compensation_HR)

When a user executes a hybrid or dense vector query via Microsoft Azure AI Foundry SDK, the cosine similarity step evaluates distance against every vector in the index. The LLM then receives retrieved context containing chunks the user lacks permissions to view. Filtering post retrieval (in memory) is insufficient because it reduces the effective k nearest neighbors returned to the LLM, causing context deprivation and potential security leaks.

Architecture: Identity Aware Filter Ingestion

To implement native RLAC in Azure AI Search, security metadata must be indexed alongside the payload and vector embeddings.

During indexing, capture Microsoft Entra ID object identifiers (OIDs) or security group IDs permitted to read the document.

Index Schema Definition (Python)

  
from azure.core.credentials import AzureKeyCredential
from azure.search.documents.indexes import SearchIndexClient
from azure.search.documents.indexes.models import (
    SearchIndex,
    SimpleField,
    SearchableField,
    SearchField,
    SearchFieldDataType,
    VectorSearch,
    HnswAlgorithmConfiguration,
    VectorSearchProfile
)


def create_rlac_vector_index(endpoint: str, credential: AzureKeyCredential, index_name: str):
    client = SearchIndexClient(endpoint=endpoint, credential=credential)
    
    fields = [
        SimpleField(name="id", type=SearchFieldDataType.String, key=True),
        SearchableField(name="content", type=SearchFieldDataType.String),
        SearchField(
            name="content_vector",
            type=SearchFieldDataType.Collection(SearchFieldDataType.Single),
            searchable=True,
            vector_search_dimensions=1536,
            vector_search_profile_name="myHnswProfile"
        ),
        # Security Filter Attributes
        SimpleField(
            name="allowed_groups",
            type=SearchFieldDataType.Collection(SearchFieldDataType.String),
            filterable=True
        ),
        SimpleField(
            name="owner_oid",
            type=SearchFieldDataType.String,
            filterable=True
        )
    ]
    
    vector_search = VectorSearch(
        algorithms=[HnswAlgorithmConfiguration(name="myHnsw")],
        profiles=[VectorSearchProfile(name="myHnswProfile", algorithm_configuration_name="myHnsw")]
    )
    
    index = SearchIndex(name=index_name, fields=fields, vector_search=vector_search)
    client.create_or_update_index(index)



  

Security Filtered Vector Retrieval Pipeline

When a client queries Azure AI Foundry, extract their Entra ID JWT payload to read their oid and groups claims. Construct an Azure AI Search OData filter expression to apply alongside the vector query.

Python

import os
from azure.identity import DefaultAzureCredential
from azure.search.documents import SearchClient
from azure.search.documents.models import VectorizedQuery
from azure.ai.evaluation import AzureAIProjectClient

def query_vector_store_with_rlac(
    search_endpoint: str,
    index_name: str,
    query_text: str,
    query_vector: list[float],
    user_oid: str,
    user_groups: list[str]
):
    credential = DefaultAzureCredential()
    search_client = SearchClient(endpoint=search_endpoint, index_name=index_name, credential=credential)
    
    # Format OData security filter expression
    groups_filter = ", ".join([f"'{g}'" for g in user_groups])
    odata_filter = f"owner_oid eq '{user_oid}' or allowed_groups/any(g: search.in(g, '{groups_filter}'))"
    
    vector_query = VectorizedQuery(
        vector=query_vector,
        k_nearest_neighbors=5,
        fields="content_vector"
    )
    
    results = search_client.search(
        search_text=query_text,
        vector_queries=[vector_query],
        filter=odata_filter,
        select=["id", "content"]
    )
    
    retrieved_chunks = [doc["content"] for doc in results]
    return retrieved_chunks
    

Performance & Optimisation Best Practices

Index Warmup & Memory Overhead: Adding string collection fields for security filtering increases index memory overhead slightly. Ensure filterable attributes are explicitly flagged with filterable=True and avoid making large security collections searchable.

Token Invalidation: When user permissions change in Microsoft Entra ID, ensure token caches expire appropriately to reflect revoked security groups.

Power BI Kids competition 2026

 Celebrating the next generation of data rockstars! 🚀✨

I am incredibly proud to share that we have officially wrapped up our 8-week #PowerBI for #Kids program this year. To celebrate their hard work, we concluded the course with an online competition where 11 talented young students showcased their skills. Thank you Medha Bandari and Tanuja Garikipati for all your help during this time.

Watching these kids transform raw data into insightful stories—using everything from Smart Narratives and Decomposition Trees to custom map visuals—was truly inspiring. 

Topics ranged from Harry Potter and Marvel movie analytics to fitness tracking and global sports stars. 📊

A huge thank you to our esteemed panel of judges:

Reza Rad (Microsoft MVP & Regional Director) – A special shout-out to Reza for joining us at 2:30 AM from Toronto! Your dedication to the community is unmatched.

Anupama Natarajan (Microsoft AI MVP)-- Thank you for accepting even though it was a last minute 

Allison Kennedy (Microsoft Data Platform MVP) -- Thank you for accepting even though it was a last minute 

#Congratulations to our winners:

🏆 1st Place: Veda(Must-visit places in India analysis)

🥈 2nd Place: Sudhendra (Novatech company performance)

🥉 3rd Place: Alvisa (Steam games player analytics)

🎗️ Leila Etaati Memorial Prize: Kumuda (Fitness and well-being analysis) — Awarded in honor of Leila Etaati, who was always a passionate supporter of these competitions.

I want to express my sincere gratitude to all the parents and kids for their commitment over the last two months. Seeing the confidence and technical growth in these students makes it all worth it! 

Interested in seeing what these kids achieved? 

Check out the recording of the competition below:

🔗 https://www.youtube.com/watch?v=nyZ3uKokNwE

Processing Nested JSON with PySpark in Microsoft Fabric

I have been working on a project in Microsoft Fabric, where I am calling an API using a pipeline and getting JSON files into the Files section of the Lakehouse. These JSON files track every insert, update, and delete that happens in the source database.

In this post, I will walk through this real-world example step by step. The pipeline copies these JSON files sitting into the Lakehouse Files folder, each containing nested CDC data. 

Our goal was to:

• Read each JSON file using PySpark inside a Fabric Notebook

• Navigate the nested structure to extract column names and their values

• Pivot those values so each column name becomes its own table column

• Add metadata columns (timestamp, start_lsn and operation type)

• Save the result as a Delta table — automatically named after the source file

• Loop through all files in the folder so the whole process is automated

By the end of this post you will have a fully parameterised, reusable notebook that you can trigger from a Fabric Pipeline.

Understanding the Source Data

Before we write any code, it helps to understand what our data looks like. Each JSON file has a structure like this:

 {
  "data": {
    "currentItemCount": 20,
    "items": [
      {
        "changes": [
          {
            "__$operation": 3,
            "data": [
              { "columnName": "PKID",  "value": "3834523" },
              { "columnName": "Name", "value": "John Doe" }
            ]
          }
        ]
      }
    ]
  }
}

When CDC files are involved the key things to notice are: 

• data.items — an array of change sets 
• items[].changes — an array of individual changes, each with an operation code 
•  __$operation — a number representing the CDC operation (1=Delete, 2=Insert, 3=Update Before, 4=Update After) 
• changes[].data — an array of key-value pairs where columnName is the field name and value is the data 

Our task is to turn those key-value pairs into proper columns in a Delta table.

Setting Up Your Fabric Notebook

Open a new Notebook in your Fabric workspace and make sure it is connected to your default Lakehouse. The notebook will be organised into clearly labelled cells.

 

💡 Tip: In Fabric Notebooks you can mark a cell as a Parameters cell by clicking the three-dot menu (...) on the cell and selecting Toggle Parameter Cell. This lets a Pipeline pass in values at runtime without changing your code.

 

Cell 1: Parameters

This cell defines the folder path where your JSON files are stored. Mark this as a Parameters cell so a Pipeline can override it.

 

# Default parameter - override this from a Fabric Pipeline

source_folder = "Files/your_folder_name"

 

💡 Tip: The source_folder value is a relative path inside the Files section of your Lakehouse. Right-click the folder in your Lakehouse explorer and choose Properties to confirm the exact path.

 

Cell 2: Imports

We need a small set of PySpark functions. Import them all upfront to keep things tidy.

 

from pyspark.sql.functions import (

    col, explode, monotonically_increasing_id,

    current_timestamp, when

)

import os

Cell 3: List the JSON Files

We use mssparkutils — a utility built into Fabric Notebooks — to list all the files in our folder. We then filter to only JSON files.

 

# List all files in the source folder

files = mssparkutils.fs.ls(f'/{source_folder}')

 

# Keep only .json files

json_files = [f for f in files if f.name.endswith('.json')]

 

print(f'Found {len(json_files)} JSON files:')

for f in json_files:

    print(f'  {f.name}')

 

💡 Tip: mssparkutils.fs.ls() returns a list of file objects. Each object has a .name property (just the filename) and a .path property (the full path you need for spark.read).

 

Cell 4: Operation Code Helper Function

The __$operation field in CDC data is a number. This helper function maps those numbers to human-readable labels, which makes the final table much easier to understand.

 

def map_operation(df_in):

    return df_in.withColumn('operation',

        when(col('operation') == 1, 'Delete')

       .when(col('operation') == 2, 'Insert')

       .when(col('operation') == 3, 'Update Before')

       .when(col('operation') == 4, 'Update After')

       .otherwise('Unknown')

    )

 

Cell 5: The Main Loop

This is the heart of the notebook. For each JSON file --

•       Derive the Delta table name from the filename

•       Read the JSON file into a Spark DataFrame

•       Explode the nested arrays step by step

•       Pivot the key-value pairs into wide-format columns

•       Add inserted_date and operation columns

•       Append the result to a Delta table

 

for file in json_files:

    # Derive table name from filename (strip .json extension)

    table_name = os.path.splitext(file.name)[0]

 

    # Sanitise: replace hyphens and spaces with underscores

    table_name_clean = table_name.replace('-', '_').replace(' ', '_')

 

    print(f'\nProcessing: {file.name} -> Table: {table_name_clean}')

 

    try:

        # Step 1: Read the JSON file

        df = spark.read.option('multiline', 'true').json(file.path)

 

        # Step 2: Explode the items array

        df_items = df.select(explode(col('data.items')).alias('item'))

 

        # Step 3: Explode changes, then select __$operation

        #         Note: backticks are required around __$operation

        #         because the $ sign is a special character in Spark

        df_changes = df_items.select(

            explode(col('item.changes')).alias('change')

        ).select(

            monotonically_increasing_id().alias('change_id'),

            col('change.`__$operation`').alias('operation'),

            col('change.data').alias('change_data')

        )

 

        # Step 4: Explode the data array (columnName + value pairs)

        df_data = df_changes.select(

            col('change_id'),

            col('operation'),

            explode(col('change_data')).alias('record')

        )

 

        # Step 5: Pull out the columnName and value fields

        df_kv = df_data.select(

            col('change_id'),

            col('operation'),

            col('record.columnName').alias('column_name'),

            col('record.value').alias('value')

        )

 

        # Step 6: Pivot so each unique columnName becomes its own column

        df_wide = df_kv.groupBy('change_id', 'operation') \

                       .pivot('column_name') \

                       .agg({'value': 'first'})

 

        # Step 7: Add metadata columns and drop the surrogate key

        df_final = df_wide \

            .withColumn('inserted_date', current_timestamp()) \

            .drop('change_id')

 

        # Step 8: Map operation codes to readable labels

        df_final = map_operation(df_final)

 

        # Step 9: Append to Delta table named after the file

        df_final.write \

            .format('delta') \

            .mode('append') \

            .option('mergeSchema', 'true') \

            .saveAsTable(table_name_clean)

 

        row_count = df_final.count()

        print(f'  Done: {row_count} rows appended to {table_name_clean}')

 

    except Exception as e:

        print(f'  ERROR processing {file.name}: {str(e)}')

        continue

 

print('\nAll files processed.')

 

Understanding Each Step in Detail

If you are new to PySpark, the explode and pivot steps can look a little daunting. Here is a plain-English explanation of what is happening at each stage.

 

What does explode() do?

When your data has an array (a list of items inside a field), explode() turns each element of that array into its own separate row. Think of it like unzipping a zip file — each item gets its own space.

In the data used, there is a need to call explode() three times because there were arrays nested inside arrays:

•       First on data.items to get one row per change set

•       Then on item.changes to get one row per individual change

•       Then on change.data to get one row per column name/value pair

 

What does pivot() do?

After exploding, our data is in a long format — each row has a column_name and a value. The pivot() function rotates these rows so that each unique column_name becomes its own column and the value moves into the corresponding cell. This is exactly the wide format you would expect in a database table.

 

Why to use backticks around __$operation?

The __$operation field name contains special characters (underscores and a dollar sign). Spark's column expression parser would misinterpret these, so wrap the name in backtick characters to tell Spark to treat the whole string as a single column name.

 

What is mergeSchema?

When appending to a Delta table, if a new JSON file introduces a new column that did not exist before, Spark would normally throw an error. The mergeSchema option tells Delta to automatically add the new column to the table schema rather than failing. This makes the notebook robust to schema changes in your source data.

 

Tips and lessons learnt

 

⚠️ 1: Do not call from_json() on a column that is already a struct. When Spark reads a JSON file, nested objects are automatically parsed into structs. Calling from_json() on a struct causes an AnalysisException because it expects a string. Navigate the struct directly using dot notation like col('data.items').

 

⚠️ 2: You cannot reference a column created by explode() in the same select() call. Always chain a second .select() after the explode so that the new column is fully resolved before you try to access its fields.

 

⚠️ 3: Column names with special characters like $ or spaces must be wrapped in backticks inside col() expressions. Without backticks, Spark will fail to parse the column name and throw an UNRESOLVED_COLUMN error.

 

⚠️ 4: Use mode('append') with option('mergeSchema', 'true') — not option('overwriteSchema', 'true'). overwriteSchema only applies when you are using overwrite mode. Using it with append mode will cause an error.

 

💡 5: Use os.path.splitext(file.name)[0] to cleanly strip the .json extension from the filename. This is more reliable than using .replace('.json', '') because splitext handles edge cases like filenames that contain the word json elsewhere.

 

Triggering from a Fabric Pipeline

Because the source_folder cell is marked as a Parameters cell, this notebook can be called from a Fabric Data Pipeline. In the Pipeline, add a Notebook activity and under the Settings tab you will find a Base parameters section where you can pass in the folder path dynamically.

 

Parameter Name : source_folder

Value          : Files/cdc_data/2024-01-15

 

This means you can have a Pipeline that processes a different date folder every day simply by passing the folder name as a parameter — no code changes needed.

 

Summary

In this post I have explained a complete, beginner-friendly solution for processing nested CDC JSON files in Microsoft Fabric. Here is a quick recap of what is covered:

•       How to use mssparkutils.fs.ls() to list files in your Lakehouse

•       How to navigate nested JSON structures using dot notation and explode()

•       Why backticks are needed for column names with special characters

•       How to pivot key-value pairs into a wide table format

•       How to add inserted_date and a human-readable operation label

•       How to append to Delta tables with automatic schema evolution using mergeSchema

•       How to parameterise the notebook so it works with any folder from a Pipeline

 

The full notebook is reusable — just drop new JSON files into the folder and run the Pipeline again. Each file gets its own Delta table named after the source file, making it easy to trace data back to its origin.