Overview
This sample demonstrates how to create a Business Activity Monitoring (BAM) transaction view for an Azure Data Factory pipeline. The scenario tracks a file-replication pipeline that processes files received from branch offices, surfacing run history and error events in a business-friendly view for support and operations teams.
Business value
This sample shows how to use BAM data queries to give support operators and business users a clear view of Data Factory pipeline runs — including timing and failure states — without requiring direct access to Azure diagnostics.
How it works
The parent transaction query retrieves the latest log record for each pipeline run and projects the columns relevant to this scenario. The stage query drills into a specific run using its correlation ID, exposing run details and a direct link to the Azure Data Factory portal for deeper investigation.
Walkthrough video
The video below walks through the setup for this scenario and highlights the key features available.
Parent transaction query
The query below retrieves the latest log entry for each run of the MoveFiles pipeline and projects the columns relevant to this scenario.
let inputSubscriptionId = "08a281b8-3b07-4219-a517-b11230e9b34f";
let inputResourceGroupName = "EAI_APP_DAILYFILEREPLICATION";
let inputDataFactory = "KV-EAI-FILEMOVES-DF";
let inputPipelineName = "MoveFiles";
//Query for the log event for the pipeline being queued to start, we can also project properties if needed
//for example | extend sourceLocation = Parameters_SourceStore_Location_s
AzureDiagnostics
| where ResourceProvider == "MICROSOFT.DATAFACTORY"
| where Category == "PipelineRuns"
| where SubscriptionId == inputSubscriptionId
| where ResourceGroup == inputResourceGroupName
| where Resource == inputDataFactory
| where pipelineName_s == inputPipelineName
//Summarize here to get the latest record for each pipeline run
| summarize arg_max(TimeGenerated, *) by CorrelationId
//Extend this column to have a friendly mappable status
| extend OverallStatus = case(
status_s == 'Succeeded', 'Succeeded',
status_s == 'Failed', 'Failed',
'In Progress')
//These columns are common to datafactory
| extend StartTime = SystemParameters_ExecutionStart_t
| extend EndTime = end_t
| extend PipelineName = pipelineName_s
| extend RunId = runId_g
| extend Duration = EndTime - StartTime
//These columns are my custom properties
| extend SourceLocation = Parameters_SourceStore_Location_s
| extend SourceDirectory = Parameters_SourceStore_Directory_s
| extend DestinationLocation = Parameters_DestinationStore_Location_s
| extend DestinationDirectory = Parameters_DestinationStore_Directory_s
//Project our just the columns I want
| project TimeGenerated, Resource, OperationName, OverallStatus, CorrelationId, RunId, PipelineName, StartTime, EndTime, SourceLocation, SourceDirectory, DestinationLocation, DestinationDirectory, Duration
| order by TimeGenerated desc
Stage query
The stage query retrieves the details for a specific pipeline run using the CorrelationId passed from the parent transaction. It also constructs a portal URL so you can navigate directly to the run in Azure Data Factory.
//Provide the input parameters for this query
let inputCorrelationId = {CorrelationId};
let inputSubscriptionId = "08a281b8-3b07-4219-a517-b11230e9b34f";
let inputResourceGroupName = "EAI_APP_DAILYFILEREPLICATION";
let inputDataFactory = "KV-EAI-FILEMOVES-DF";
let inputPipelineName = "MoveFiles";
//Query for the log event for the pipeline being queued to start, we can also project properties if needed
//for example | extend sourceLocation = Parameters_SourceStore_Location_s
AzureDiagnostics
| where ResourceProvider == "MICROSOFT.DATAFACTORY"
| where Category == "PipelineRuns"
| where SubscriptionId == inputSubscriptionId
| where ResourceGroup == inputResourceGroupName
| where Resource == inputDataFactory
| where pipelineName_s == inputPipelineName
| where CorrelationId == inputCorrelationId
//Get the most recent record which will be success or failure
| summarize arg_max(TimeGenerated, *) by CorrelationId
//Map the status
| extend OverallStatus = case(
status_s == 'Succeeded', 'Succeeded',
status_s == 'Failed', 'Failed',
'In Progress')
//This will create a url we can use for accessing via the azure portal
| extend PortalUrl = strcat("https://adf.azure.com/en/monitoring/pipelineruns/", runId_g, "?factory=%2Fsubscriptions%2F", inputSubscriptionId, "%2FresourceGroups%2F", inputResourceGroupName, "%2Fproviders%2FMicrosoft.DataFactory%2Ffactories%2F", inputDataFactory)