ADF Categories

Move and transform Activities

Explore the building blocks of Azure Data Factory pipelines within this category.

Tutorial Modules

Pipeline JSON Breakdown

Usecase1: Filesystems Ingestion

Objective: Copy files from Azure Blob Storage → ADLS Gen2, validate file existence, handle failures gracefully, and delete source files after successful ingestion.

Pipeline Flow Architecture

Interactive diagram based on the configured dependencies.

Click to Maximize

Implementation Steps

Step 1 & 2: Create Linked Services & Datasets

Before creating activities, we need to define the connections and datasets referenced in the pipeline.

Source Dataset: SourceDataset_yun (Azure Blob, Delimited Text)

Sink Dataset: DelimitedText1 (ADLS Gen2, Delimited Text)

Staging: CTSJobADLSTarget1

Step 3: Add Validation Activity

Drag Validation Activity to the canvas. This ensures the file exists before processing.

NameValidation1

DatasetSourceDataset_yun

Timeout0.00:02:00

Sleep10 seconds

Step 4: Add Fail Activity (Validation Failure)

Connect from Validation → On Failure path.

Name: Fail1_copy1

Message: "source file (cust_header.csv) is not present for the given run"

Error Code: "1 : filenotfound"

Step 5: Add Copy Activity

Connect from Validation → On Success path.

Source Configuration

Type: DelimitedTextSource
Recursive: True
Wildcard: Folder: *, File: *

Sink Configuration

Type: DelimitedTextSink
Copy Behavior: PreserveHierarchy
File extension: .txt
Quote All Text: True

Additional Settings

Parallel Copies: 2

DIUs: 4

Staging: Enabled (Linked Service: CTSJobADLSTarget1)

Logging: Enabled (Warning Level)

Step 6: Add Fail Activity (Copy Failure)

Connect from Copy Data → On Failure path.

Name: Fail1

Message: "some issue in the data copy though source file is present"

Error Code: "2:somecopyissue"

Step 7: Add Delete Activity

Connect from Copy Data → On Success path. Cleans up source files after successful ingestion.

Name: Delete1

Dataset: SourceDataset_yun

Recursive: True

Logging: False

Step 8: Debug & Publish

Click Debug to run the pipeline, verifying validation, copy, and delete operations. Once verified, click Publish to deploy.

Pipeline JSON Breakdown

Usecase2: DB (SOI to Datalake) Ingestion

Objective: Extract data directly from a MySQL Database and sink it to Azure Data Lake Storage as structured JSON output, flattening the hierarchy automatically.

Pipeline Flow Architecture

Interactive diagram based on the configured dependencies.

Click to Maximize

Implementation Steps

Step 1: Create Linked Services

Navigate to Manage → Linked Services → + New.

Source: MySQL Database

Name: LS_MySQL_Source
Connect via IR: AutoResolveIntegrationRuntime
Configuration: Provide Server name, Port, Database name, and Auth details.
Test Connection: Ensure it is successful.

Sink: Azure Data Lake Storage Gen2

Name: LS_ADLS_Sink
Auth method: Account Key (or Service Principal)
Storage account: Select from dropdown.
Test Connection: Ensure it is successful.

Step 2: Configure Datasets

Navigate to Author → Datasets → + New Dataset.

Source Dataset (MySQL)

Type: MySQL
Name: MySqlTable1
Linked service: Select LS_MySQL_Source
Table name: (Leave blank for custom query)

Sink Dataset (ADLS Gen2)

Type: ADLS Gen2 → JSON
Name: DestinationDataset_yun
Linked service: Select LS_ADLS_Sink
File path: container/folder/output.json

Step 3: Pipeline Configuration

Create a new Pipeline named Usecase2_DB_Ingestion. Drag the Copy Data activity onto the canvas.

General & Source Tabs

General - Name: Ingest SOR DB Data to Datalake
Source - Dataset: MySqlTable1
Source - Use query: Query

Query text:

select * from telco_schema.dim_tower_source where network_type='5G'

Sink Tab

Sink dataset: DestinationDataset_yun
File pattern: arrayOfObjects
Copy behavior: FlattenHierarchy

Settings Tab

Enable staging: Unchecked (False)

Policy Section

Timeout: 0.12:00:00 (12 hours)
Retry: 1
Retry interval (sec): 30

Step 4: Validate and Publish

Validate to check for configuration errors.

Debug to run a test execution.

Publish All to save changes.

Pipeline JSON Breakdown

Usecase3: SAP (OData to Datalake) Ingestion

Objective: Extract data from an SAP System via OData using a filtered query and land it as structured JSON in Azure Data Lake Storage Gen2.

Pipeline Flow Architecture

Interactive diagram based on the configured dependencies.

Click to Maximize

Implementation Steps

Step 1: Create Linked Services

Navigate to Manage → Linked Services → + New.

Source: SAP OData

Type: OData
Name: LS_SAP_OData
Service URL: OData endpoint URL from SAP
Auth type: Basic or appropriately configured Auth

Sink: Azure Data Lake Gen2

Type: Azure Data Lake Storage Gen2
Name: LS_ADLS_Sink

Step 2: Configure Datasets

Navigate to Author → Datasets → + New Dataset.

Source Dataset (OData)

Type: OData
Name: ODataResource1
Linked service: Select LS_SAP_OData

Sink Dataset (ADLS JSON)

Type: ADLS Gen2 → JSON
Name: Json1
Linked service: Select LS_ADLS_Sink

Step 3: Pipeline Configuration

Create a new Pipeline named usecase3_SAP_ingestion. Drag the Copy Data activity onto the canvas.

General & Source Tabs

General - Name: Ingesting_SAP_Items
Source - Dataset: ODataResource1
Http Request Timeout: 00:05:00 (5 minutes)

OData Query:

$select=ObjectID,ParentObjectID,TypeCode&filter=Typecode eq '002'

Sink Tab

Sink dataset: Json1
Copy behavior: FlattenHierarchy

Settings Tab

Enable staging: Unchecked (False)

Policy Section

Timeout: 0.12:00:00 (12 hours)
Retry: 0

Step 4: Validate and Publish

Pipeline JSON Breakdown

Usecase4: Multicloud Lakehouse Ingestion

Objective: Extract high-value claim transactions directly from Google BigQuery using a custom SQL query and ingest them into Azure Data Lake Storage Gen2 as structured JSON.

Pipeline Flow Architecture

Interactive diagram based on the configured dependencies.

Click to Maximize

Implementation Steps

Step 1: Create Linked Services

Navigate to Manage → Linked Services → + New.

Source: Google BigQuery

Type: Google BigQuery
Name: LS_GoogleBigQuery
Auth type: Service Account / User Authentication
Project ID: Your GCP Project ID

Sink: Azure Data Lake Gen2

Type: Azure Data Lake Storage Gen2
Name: LS_ADLS_Sink

Step 2: Configure Datasets

Navigate to Author → Datasets → + New Dataset.

Source Dataset (BigQuery)

Type: Google BigQuery
Name: GoogleBigQueryObject1
Linked service: Select LS_GoogleBigQuery

Sink Dataset (ADLS JSON)

Type: ADLS Gen2 → JSON
Name: Json1
Linked service: Select LS_ADLS_Sink

Step 3: Pipeline Configuration

Create a new Pipeline named Usecase4_Multicloud_Lakehouse_Ingestion. Drag the Copy Data activity onto the canvas.

General & Source Tabs

General - Name: Copy data1
Source - Dataset: GoogleBigQueryObject1
Source Type: GoogleBigQueryV2Source

SQL Query:

select * from inceptez_insurance.claim_transactions where claim_amount>1000

Sink Tab

Sink dataset: Json1
Copy behavior: PreserveHierarchy
File pattern: arrayOfObjects

Sink Settings Tab

Max concurrent connections: 2
Block size in MB: 50
Enable staging: Unchecked (False)

Policy Section

Timeout: 0.12:00:00 (12 hours)
Retry: 0

Step 4: Validate and Publish

Pipeline JSON Breakdown

Usecase5: REST API & ForEach Iteration

Objective: Orchestrate a dynamic data pipeline using variables to iterate over an array of currencies, calling a live Exchange Rate REST API, and enriching database records before sinking to Delimited Text.

Pipeline Flow Architecture

Interactive diagram based on the configured dependencies.

Click to Maximize

Implementation Steps

Step 1: Pipeline Parameters & Variables

Initialize the parameters and variables to store dynamic inputs.

Parameters

currencyparam1: Array [USD, INR, AED]
filename: String "CurrData"

Variables

currency_list: Array
all_data, webarray: Array

Step 2: Control Flow Init (Set Variable)

Assign the parameter array to a variable and handle success/failure routing.

Set variable1

Variable: currency_list
Value: @pipeline().parameters.currencyparam1

Fail1 (On Failure)

Message: failure occur in setting activity

Wait2 (On Success)

Wait Time: 1 Second

Step 3: Setup ForEach Iteration

Drag a ForEach activity, link it after Wait2, and configure the items it will loop through.

Name: ForEach1
Sequential: True
Items (Expression): @variables('currency_list')

Step 4: Inside ForEach Loop (Web & Copy Data)

Inside the ForEach canvas, build the dynamic activities that run for each currency iteration.

1. Web1 (REST API Call)

Method: GET
URL: @concat('https://.../latest/', item())

Fail2 (On Failure): '@concat('Failed for this currency ',item())'

2. Wait1 & 3. Copy data2

Wait Time: 1 Second
Copy Source (MySQL): select 1 as dummy
Additional Columns Mapped:
- currency = @string(item())
- webjson = @string(activity('Web1').output)

Step 5: Validate and Publish

Pipeline JSON Breakdown

Usecase6: Advanced SQL Ingestion Patterns

Objective: Master various relational database loading strategies including Truncate & Load, Upsert/Merge, CDC Watermarking, and conditional control flows using Lookups and If Conditions.

Pipeline Flow Architecture

Interactive diagram based on the conditional dependencies (CDC flow).

Click to Maximize

Implementation Steps

Step 1: Simple Load Patterns

Review the basic patterns for loading data into an Azure SQL Sink.

Truncate and Load

Write Behavior: Insert
Pre-copy Script: truncate table izdb.cdr;
Clears the table entirely before loading new rows.

Merge or Upsert Load

Write Behavior: Upsert
Upsert Settings: Use Temp DB = True
Key Columns: ['cid']

Step 2: Conditional Control Flow

Orchestrating logic based on source database states using lookups.

1. Validation1

Validates AzureSqlTable3 exists
Sleep: 10 sec

2. Lookup1

Query: select count(1) as cnt from cust_dim
Depends on Validation Succeeded.

3. If Condition1

Expression:
@greater(int(activity('Lookup1').output.firstRow.cnt), 0)

Step 3: CDC Watermarking (If True Branch)

If data exists, find the latest timestamp and extract only newer records.

Watermark Lookup

Type: Azure SQL Source
SQL Query:
select coalesce(max(updateddt),'2020-01-01') as watermark...

CDC Copy Activity

Dynamic Query:
@concat('select * from telco_schema.cust_dim where updateddt>''', formatDateTime(activity('watermark max...').output.firstRow.watermark, 'yyyy-MM-dd HH:mm:ss'),'''')
Sink: Upsert based on custid

Step 4: Incremental Fact Loading via Parameters

Alternatively, use a pipeline parameter to drive data ingestion.

FACT_Incremental_SOR_SOR Activity

Source Query: @concat('select * from txn_fact where loaddt>=''',pipeline().parameters.loaddt,'''')
Pre-copy Script: @concat('delete from izdb.txn_fact where loaddt>=''',pipeline().parameters.loaddt,'''')

Step 5: Validate and Publish

Pipeline JSON Breakdown

Usecase7: Databricks Medallion ETL

Objective: Ingest data from Google BigQuery to ADLS as Parquet, then orchestrate a Medallion architecture (Silver and Gold layers) using connected Azure Databricks Notebooks, including automated failure cleanup.

Pipeline Flow Architecture

Interactive diagram based on the configured dependencies.

Click to Maximize

Implementation Steps

Step 1 & 2: Configure Datasets & Computes

Prepare the source, sink, and compute connections needed for the Medallion flow.

Source Dataset

Type: Google BigQuery
Name: GoogleBigQueryObject2

Sink Dataset

Type: ADLS Gen2 → Parquet
Name: Parquet1

Compute

Linked Service: AzureDatabricks2

Step 3: Initial Ingestion (Bronze)

Ingest raw data into the Bronze layer as Parquet.

Name: Copy data1
Source: GoogleBigQueryV2Source
Sink: ParquetSink (FlattenHierarchy)

Step 4: Execute Databricks Notebooks (Silver & Gold)

Trigger the Databricks Spark notebooks sequentially upon successful ingestion.

Load Silver

Type: DatabricksNotebook
Depends On: Copy data1 (Succeeded)
Notebook Path:
/.../11_Datafactory_Usecases/1_load_silver

Load Gold

Type: DatabricksNotebook
Depends On: Load Silver (Succeeded)
Notebook Path:
/.../11_Datafactory_Usecases/2_load_gold

Step 5: Automated Cleanup on Failure

If the Silver processing fails, delete the raw Parquet file to maintain idempotency.

Activity: Delete1
Depends On: Load Silver (Failed)
Target Dataset: Parquet1

Step 6: Validate and Publish

Pipeline JSON Breakdown

Usecase8: Pipeline Orchestration & Databricks

Objective: Master master-worker pipeline orchestration by using the Execute Pipeline activity to invoke a REST API ingestion pipeline, passing parameters, and subsequently triggering a Databricks Notebook for data curation.

Pipeline Flow Architecture

Interactive diagram based on the configured dependencies.

Click to Maximize

Implementation Steps

Step 1: Define Pipeline Parameters

Create the master pipeline parameters that will be passed down to the child pipeline and Databricks.

Master Pipeline Parameters

currencyparam1 (Array): ["USD", "INR", "AED"]
filename (String): "SalesData"

Step 2: Add Execute Pipeline Activity

Use the Execute Pipeline activity to trigger the REST API ingestion pipeline built in Usecase 5.

General Configuration

Name: API Ingestion Pipeline
Invoked Pipeline: Usecase5_REST_API
Wait on completion: True

Parameters to Pass

currencyparam1: @pipeline().parameters.currencyparam1
filename: "CurrencyData"

Step 3: Curate Data with Databricks

Once the ingestion pipeline completes successfully, trigger the Databricks notebook for JSON curation.

Databricks Notebook Activity

Name: API JSON Curation
Depends On: API Ingestion Pipeline (Succeeded)
Linked Service: AzureDatabricks2
Notebook Path: /.../ADF_usecase_8_API_CALL
Base Parameters: filename = @pipeline().parameters.filename

Step 4: Validate and Publish

Pipeline & Dataflow JSON Breakdown

Usecase9: Mapping Data Flow with Parameters

Objective: Pass dynamic pipeline parameters into a Mapping Data Flow to partition data. Transform the source data stream using branches, ranks, aggregations, and row-alteration policies before writing to multiple sinks.

Mapping Data Flow Architecture

Interactive diagram based on the configured Data Flow transformations and branches.

Click to Maximize

Implementation Steps

Step 1: Pipeline Parameters & Execute Data Flow

Configure the main pipeline to pass parameters into the Data Flow engine.

Pipeline Setup

Pipeline Parameter: pipelineparam1 (int) = 4
Activity: Execute Data Flow (Data flow1)
Compute: General, 8 Cores

Passing Parameters

Dataflow Parameter: partitionparam
Expression Value:
@pipeline().parameters.pipelineparam1

Step 2: Data Flow Source & Filtering

Extract data from Azure SQL and apply initial row-level filtering.

Source1

Dataset: AzureSqlTable1
Partitioning: Round Robin using $partitionparam
Enable CDC: True

Filter1

Incoming Stream: source1
Expression: greaterOrEqual(call_dur, 401)

Step 3: Branch 1 - Schema Modification & File Sink

Transform the data structure and write to a delimited text file.

1. Select (select1): Rename custid to customerid.
2. Derived Column (derivedColumn1): Add new column loaddt = currentDate().
3. Surrogate Key (surrogateKey1): Generate seqnosk starting at 1.
4. Cast (cast1): Ensure customerid is cast to String type.
5. Sink (sink1): Output to DelimitedText2.

Step 4: Branch 2 - Aggregation & DB Upsert

Calculate aggregates and conditionally modify rows before updating the SQL table.

1. Rank (rank1): Generate custid_rnk by sorting call_dur descending.
2. Aggregate (aggregate1): Group by custid and custid_rnk.
Calculate avg_call_dur = avg(call_dur) and sum_cost = sum(cost).
3. Alter Row (alterRow1): Policy: deleteIf(lesserOrEqual(sum_cost,5)).
4. Sink (sink2): Output to AzureSqlTable4. Enable Insert, Update, Delete, Upsert policies matching custid.