April 15, 2021 6 min read

Azure Data Factory Triggers - Orchestrating Data Pipelines

Azure Data Factory ETL Data Orchestration Automation

Azure Data Factory triggers are the heartbeat of your data orchestration. They determine when and how your pipelines execute. Understanding the different trigger types and their configurations is crucial for building reliable data workflows. Let me walk you through the options and best practices.

Types of Triggers

ADF supports four trigger types:

Schedule Trigger - Time-based execution
Tumbling Window Trigger - Time-sliced processing with dependencies
Event Trigger - Blob storage events
Custom Event Trigger - Event Grid custom events

Schedule Triggers

Basic Schedule Trigger

{
  "name": "DailyETLTrigger",
  "type": "ScheduleTrigger",
  "typeProperties": {
    "recurrence": {
      "frequency": "Day",
      "interval": 1,
      "startTime": "2021-04-15T02:00:00Z",
      "endTime": "2022-04-15T02:00:00Z",
      "timeZone": "UTC"
    }
  },
  "pipelines": [
    {
      "pipelineReference": {
        "referenceName": "DailyDataIngestion",
        "type": "PipelineReference"
      },
      "parameters": {
        "ProcessDate": "@trigger().scheduledTime"
      }
    }
  ]
}

Complex Schedule Patterns

{
  "name": "BusinessHoursETL",
  "type": "ScheduleTrigger",
  "typeProperties": {
    "recurrence": {
      "frequency": "Week",
      "interval": 1,
      "startTime": "2021-04-15T09:00:00",
      "timeZone": "Eastern Standard Time",
      "schedule": {
        "hours": [9, 12, 15, 18],
        "minutes": [0],
        "weekDays": ["Monday", "Tuesday", "Wednesday", "Thursday", "Friday"]
      }
    }
  }
}

Multiple Pipelines from One Trigger

{
  "name": "MorningBatchTrigger",
  "type": "ScheduleTrigger",
  "typeProperties": {
    "recurrence": {
      "frequency": "Day",
      "interval": 1,
      "startTime": "2021-04-15T06:00:00Z",
      "timeZone": "UTC"
    }
  },
  "pipelines": [
    {
      "pipelineReference": {
        "referenceName": "SalesDataPipeline",
        "type": "PipelineReference"
      }
    },
    {
      "pipelineReference": {
        "referenceName": "InventoryDataPipeline",
        "type": "PipelineReference"
      }
    },
    {
      "pipelineReference": {
        "referenceName": "CustomerDataPipeline",
        "type": "PipelineReference"
      }
    }
  ]
}

Tumbling Window Triggers

Tumbling windows provide non-overlapping, contiguous time intervals - perfect for processing data in time slices.

Basic Tumbling Window

{
  "name": "HourlyProcessingTrigger",
  "type": "TumblingWindowTrigger",
  "typeProperties": {
    "frequency": "Hour",
    "interval": 1,
    "startTime": "2021-04-15T00:00:00Z",
    "delay": "00:15:00",
    "maxConcurrency": 10,
    "retryPolicy": {
      "count": 3,
      "intervalInSeconds": 30
    }
  },
  "pipeline": {
    "pipelineReference": {
      "referenceName": "HourlyAggregation",
      "type": "PipelineReference"
    },
    "parameters": {
      "WindowStart": "@trigger().outputs.windowStartTime",
      "WindowEnd": "@trigger().outputs.windowEndTime"
    }
  }
}

Tumbling Window with Dependencies

{
  "name": "DailyReportTrigger",
  "type": "TumblingWindowTrigger",
  "typeProperties": {
    "frequency": "Day",
    "interval": 1,
    "startTime": "2021-04-15T00:00:00Z",
    "delay": "02:00:00",
    "maxConcurrency": 1,
    "dependsOn": [
      {
        "type": "TumblingWindowTriggerDependencyReference",
        "referenceTrigger": {
          "referenceName": "HourlyProcessingTrigger",
          "type": "TriggerReference"
        },
        "offset": "-1.00:00:00",
        "size": "1.00:00:00"
      }
    ]
  }
}

Self-Dependency (Sequential Processing)

{
  "name": "SequentialBatchTrigger",
  "type": "TumblingWindowTrigger",
  "typeProperties": {
    "frequency": "Hour",
    "interval": 1,
    "startTime": "2021-04-15T00:00:00Z",
    "maxConcurrency": 1,
    "dependsOn": [
      {
        "type": "SelfDependencyTumblingWindowTriggerReference",
        "offset": "-01:00:00",
        "size": "01:00:00"
      }
    ]
  }
}

Event Triggers (Blob Storage)

Blob Created Trigger

{
  "name": "NewFilesTrigger",
  "type": "BlobEventsTrigger",
  "typeProperties": {
    "blobPathBeginsWith": "/raw/incoming/",
    "blobPathEndsWith": ".csv",
    "ignoreEmptyBlobs": true,
    "scope": "/subscriptions/{sub-id}/resourceGroups/{rg}/providers/Microsoft.Storage/storageAccounts/{storage-account}",
    "events": ["Microsoft.Storage.BlobCreated"]
  },
  "pipelines": [
    {
      "pipelineReference": {
        "referenceName": "ProcessNewFile",
        "type": "PipelineReference"
      },
      "parameters": {
        "FileName": "@triggerBody().fileName",
        "FolderPath": "@triggerBody().folderPath"
      }
    }
  ]
}

Blob Deleted Trigger

{
  "name": "FileDeletedTrigger",
  "type": "BlobEventsTrigger",
  "typeProperties": {
    "blobPathBeginsWith": "/processed/",
    "scope": "/subscriptions/{sub-id}/resourceGroups/{rg}/providers/Microsoft.Storage/storageAccounts/{storage-account}",
    "events": ["Microsoft.Storage.BlobDeleted"]
  },
  "pipelines": [
    {
      "pipelineReference": {
        "referenceName": "HandleDeletedFile",
        "type": "PipelineReference"
      }
    }
  ]
}

Custom Event Triggers

For integration with Event Grid topics:

{
  "name": "OrderEventTrigger",
  "type": "CustomEventsTrigger",
  "typeProperties": {
    "scope": "/subscriptions/{sub-id}/resourceGroups/{rg}/providers/Microsoft.EventGrid/topics/order-events",
    "events": [
      {
        "eventType": "Orders.OrderCreated"
      },
      {
        "eventType": "Orders.OrderUpdated"
      }
    ],
    "subjectBeginsWith": "orders/",
    "subjectEndsWith": ""
  },
  "pipelines": [
    {
      "pipelineReference": {
        "referenceName": "ProcessOrderEvent",
        "type": "PipelineReference"
      },
      "parameters": {
        "EventType": "@triggerBody().eventType",
        "EventData": "@triggerBody().data"
      }
    }
  ]
}

Pipeline Parameters from Triggers

Accessing Trigger Information

{
  "pipeline": {
    "parameters": {
      "TriggerName": "@trigger().name",
      "TriggerTime": "@trigger().startTime",
      "ScheduledTime": "@trigger().scheduledTime",
      "TriggerType": "@trigger().type"
    }
  }
}

Tumbling Window Parameters

{
  "parameters": {
    "WindowStart": "@trigger().outputs.windowStartTime",
    "WindowEnd": "@trigger().outputs.windowEndTime",
    "WindowStartFormatted": "@formatDateTime(trigger().outputs.windowStartTime, 'yyyy-MM-dd')"
  }
}

Blob Event Parameters

{
  "parameters": {
    "FileName": "@triggerBody().fileName",
    "FolderPath": "@triggerBody().folderPath",
    "FileUrl": "@triggerBody().fileUrl",
    "ContentType": "@triggerBody().contentType",
    "EventType": "@triggerBody().eventType"
  }
}

Managing Triggers with PowerShell

# Get trigger
$trigger = Get-AzDataFactoryV2Trigger `
    -ResourceGroupName "myResourceGroup" `
    -DataFactoryName "myDataFactory" `
    -TriggerName "DailyETLTrigger"

# Start trigger
Start-AzDataFactoryV2Trigger `
    -ResourceGroupName "myResourceGroup" `
    -DataFactoryName "myDataFactory" `
    -TriggerName "DailyETLTrigger"

# Stop trigger
Stop-AzDataFactoryV2Trigger `
    -ResourceGroupName "myResourceGroup" `
    -DataFactoryName "myDataFactory" `
    -TriggerName "DailyETLTrigger"

# Get trigger runs
Get-AzDataFactoryV2TriggerRun `
    -ResourceGroupName "myResourceGroup" `
    -DataFactoryName "myDataFactory" `
    -TriggerName "DailyETLTrigger" `
    -TriggerRunStartedAfter "2021-04-01" `
    -TriggerRunStartedBefore "2021-04-15"

Managing Triggers with REST API

import requests
from azure.identity import DefaultAzureCredential

credential = DefaultAzureCredential()
token = credential.get_token("https://management.azure.com/.default")

headers = {
    "Authorization": f"Bearer {token.token}",
    "Content-Type": "application/json"
}

base_url = f"https://management.azure.com/subscriptions/{subscription_id}/resourceGroups/{resource_group}/providers/Microsoft.DataFactory/factories/{factory_name}"

# Create or update trigger
trigger_definition = {
    "properties": {
        "type": "ScheduleTrigger",
        "typeProperties": {
            "recurrence": {
                "frequency": "Day",
                "interval": 1,
                "startTime": "2021-04-15T02:00:00Z"
            }
        },
        "pipelines": [
            {
                "pipelineReference": {
                    "referenceName": "MyPipeline",
                    "type": "PipelineReference"
                }
            }
        ]
    }
}

response = requests.put(
    f"{base_url}/triggers/MyTrigger?api-version=2018-06-01",
    headers=headers,
    json=trigger_definition
)

# Start trigger
response = requests.post(
    f"{base_url}/triggers/MyTrigger/start?api-version=2018-06-01",
    headers=headers
)

# Query trigger runs
query = {
    "lastUpdatedAfter": "2021-04-01T00:00:00Z",
    "lastUpdatedBefore": "2021-04-15T00:00:00Z",
    "filters": [
        {
            "operand": "TriggerName",
            "operator": "Equals",
            "values": ["MyTrigger"]
        }
    ]
}

response = requests.post(
    f"{base_url}/queryTriggerRuns?api-version=2018-06-01",
    headers=headers,
    json=query
)

trigger_runs = response.json()["value"]

Backfill and Rerun Patterns

Tumbling Window Backfill

def backfill_tumbling_window(factory_client, factory_name, trigger_name, start_time, end_time):
    """Rerun tumbling window trigger for historical data"""

    # Get trigger runs that need reprocessing
    filter_params = {
        "lastUpdatedAfter": start_time.isoformat(),
        "lastUpdatedBefore": end_time.isoformat(),
        "filters": [
            {"operand": "TriggerName", "operator": "Equals", "values": [trigger_name]},
            {"operand": "Status", "operator": "Equals", "values": ["Failed"]}
        ]
    }

    runs = factory_client.trigger_runs.query_by_factory(
        resource_group_name, factory_name, filter_params
    )

    for run in runs.value:
        # Rerun the failed window
        factory_client.trigger_runs.rerun(
            resource_group_name,
            factory_name,
            trigger_name,
            run.run_id
        )
        print(f"Rerun initiated for window: {run.trigger_run_timestamp}")

Monitoring Trigger Execution

Azure Monitor Alerts

{
  "type": "Microsoft.Insights/metricAlerts",
  "properties": {
    "description": "Alert when trigger fails",
    "severity": 2,
    "enabled": true,
    "scopes": ["/subscriptions/{sub}/resourceGroups/{rg}/providers/Microsoft.DataFactory/factories/{factory}"],
    "evaluationFrequency": "PT5M",
    "windowSize": "PT5M",
    "criteria": {
      "odata.type": "Microsoft.Azure.Monitor.MultipleResourceMultipleMetricCriteria",
      "allOf": [
        {
          "name": "TriggerFailedRuns",
          "metricName": "TriggerFailedRuns",
          "operator": "GreaterThan",
          "threshold": 0,
          "timeAggregation": "Total"
        }
      ]
    }
  }
}

Best Practices

Use appropriate trigger types - Schedule for regular batches, tumbling window for time-sliced processing, events for real-time
Set delays wisely - Allow time for source data to be complete
Configure retry policies - Handle transient failures
Monitor trigger runs - Set up alerts for failures
Use parameters - Pass trigger context to pipelines
Test thoroughly - Verify trigger configurations before production
Document dependencies - Clear understanding of trigger chains

Conclusion

Azure Data Factory triggers provide flexible orchestration options for your data pipelines. By understanding the differences between schedule, tumbling window, and event triggers, you can design robust data workflows that meet your timing and dependency requirements. The key is choosing the right trigger type for your use case and configuring it properly for reliability.