azure_data_factory_blob_copy_pipeline_quickstart.py

1import time
2from azure.identity import DefaultAzureCredential
3from azure.mgmt.datafactory import DataFactoryManagementClient
4from azure.mgmt.datafactory.models import (
5    Factory, PipelineResource, LinkedServiceResource, 
6    AzureBlobStorageLinkedService, DatasetResource, 
7    AzureBlobDataset, CopyActivity, BlobSource, 
8    BlobSink, DatasetReference, LinkedServiceReference
9)
10from azure.mgmt.resource import ResourceManagementClient
11
12# 1. Provide your configuration details
13subscription_id = 'YOUR_SUBSCRIPTION_ID'
14resource_group_name = 'ADFQuickStartRG'
15factory_name = 'ADFQuickStartFactory'
16location = 'East US'
17storage_account_name = 'YOUR_STORAGE_ACCOUNT_NAME'
18storage_account_key = 'YOUR_STORAGE_ACCOUNT_KEY'
19input_blob_path = 'containername/input'
20output_blob_path = 'containername/output'
21
22# 2. Authenticate and initialize clients
23credential = DefaultAzureCredential()
24resource_client = ResourceManagementClient(credential, subscription_id)
25adf_client = DataFactoryManagementClient(credential, subscription_id)
26
27# 3. Create a Resource Group (Optional if already exists)
28resource_client.resource_groups.create_or_update(resource_group_name, {'location': location})
29
30# 4. Create a Data Factory
31print(f"Creating Data Factory: {factory_name}")
32factory_resource = Factory(location=location)
33adf_client.factories.create_or_update(resource_group_name, factory_name, factory_resource)
34
35# 5. Create an Azure Storage Linked Service
36print("Creating Linked Service...")
37storage_ls_name = 'AzureStorageLinkedService'
38storage_ls_resource = LinkedServiceResource(
39    properties=AzureBlobStorageLinkedService(
40        connection_string=f"DefaultEndpointsProtocol=https;AccountName={storage_account_name};AccountKey={storage_account_key};EndpointSuffix=core.windows.net"
41    )
42)
43adf_client.linked_services.create_or_update(resource_group_name, factory_name, storage_ls_name, storage_ls_resource)
44
45# 6. Create Datasets (Input and Output)
46print("Creating Datasets...")
47input_dataset_name = 'InputDataset'
48input_dataset = DatasetResource(
49    properties=AzureBlobDataset(
50        linked_service_name=LinkedServiceReference(reference_name=storage_ls_name),
51        folder_path=input_blob_path
52    )
53)
54adf_client.datasets.create_or_update(resource_group_name, factory_name, input_dataset_name, input_dataset)
55
56output_dataset_name = 'OutputDataset'
57output_dataset = DatasetResource(
58    properties=AzureBlobDataset(
59        linked_service_name=LinkedServiceReference(reference_name=storage_ls_name),
60        folder_path=output_blob_path
61    )
62)
63adf_client.datasets.create_or_update(resource_group_name, factory_name, output_dataset_name, output_dataset)
64
65# 7. Create a Pipeline with a Copy Activity
66print("Creating Pipeline...")
67pipeline_name = 'CopyPipeline'
68copy_activity = CopyActivity(
69    name='CopyFromBlobToBlob',
70    inputs=[DatasetReference(reference_name=input_dataset_name)],
71    outputs=[DatasetReference(reference_name=output_dataset_name)],
72    source=BlobSource(),
73    sink=BlobSink()
74)
75pipeline_resource = PipelineResource(activities=[copy_activity])
76adf_client.pipelines.create_or_update(resource_group_name, factory_name, pipeline_name, pipeline_resource)
77
78# 8. Create a Pipeline Run
79print("Starting Pipeline Run...")
80run_response = adf_client.pipelines.create_run(resource_group_name, factory_name, pipeline_name)
81
82# 9. Monitor the Pipeline Run
83while True:
84    pipeline_run = adf_client.pipeline_runs.get(resource_group_name, factory_name, run_response.run_id)
85    print(f"Current Status: {pipeline_run.status}")
86    if pipeline_run.status in ['Succeeded', 'Failed', 'Cancelled']:
87        break
88    time.sleep(10)
89
90print("Quickstart completed.")