Session 22: Azure Data Factory & Data Integration

Mục tiêu: Master ADF pipelines, data flows, và build production ETL/ELT pipelines.

1. ADF Architecture

┌────────────────────────────────────────────────────────────────┐
│                    Azure Data Factory                           │
│                                                                 │
│  ┌──────────┐   ┌──────────┐   ┌──────────┐   ┌──────────┐  │
│  │ Pipeline │──→│ Activity │──→│ Activity │──→│ Activity │  │
│  │          │   │ (Copy)   │   │ (DataFlow)│   │ (Stored  │  │
│  │          │   │          │   │          │   │  Proc)   │  │
│  └──────────┘   └──────────┘   └──────────┘   └──────────┘  │
│       │                                                       │
│  ┌────▼─────┐           ┌──────────────┐                     │
│  │ Trigger  │           │ Integration  │                     │
│  │ (Schedule│           │ Runtime      │                     │
│  │  Event)  │           │ (Azure/Self) │                     │
│  └──────────┘           └──────────────┘                     │
│                                                               │
│  ┌──────────────┐   ┌──────────────┐                         │
│  │ Linked       │   │ Datasets     │                         │
│  │ Services     │   │ (Schema)     │                         │
│  │ (Connections)│   │              │                         │
│  └──────────────┘   └──────────────┘                         │
└────────────────────────────────────────────────────────────────┘

Key Concepts

Pipeline: container cho activities (workflow)
Activity: unit of work (Copy, Data Flow, Stored Proc, Azure Function, Web)
Trigger: khởi động pipeline (Schedule, Event, Tumbling Window)
Linked Service: connection string đến data source
Dataset: schema + reference đến data trong linked service
Integration Runtime:
  - Azure IR: managed, auto-scale
  - Self-hosted IR: on-premises data access
  - SSIS IR: lift-and-shift SSIS packages

2. Common Activities

Copy Activity (ETL: Extract → Load)

json
// Copy Data: SQL Server → Azure SQL
{
  "name": "CopyOrdersToDataWarehouse",
  "type": "Copy",
  "inputs": [
    { "referenceName": "OnPremSqlOrders", "type": "DatasetReference" }
  ],
  "outputs": [
    { "referenceName": "AzureSqlDW_Orders", "type": "DatasetReference" }
  ],
  "typeProperties": {
    "source": {
      "type": "SqlSource",
      "sqlReaderQuery": "SELECT * FROM Orders WHERE ModifiedDate > @{pipeline().parameters.LastRunTime}"
    },
    "sink": {
      "type": "AzureSqlSink",
      "writeBehavior": "upsert",
      "upsertSettings": {
        "useTempDB": true,
        "keys": ["OrderId"]
      }
    },
    "enableStaging": true,
    "stagingSettings": {
      "linkedServiceName": { "referenceName": "AzureBlobStaging" }
    }
  }
}

Data Flow (ELT: Transform in Spark)

Source → Filter → Derived Column → Aggregate → Join → Sink

Ví dụ: Financial Reporting Pipeline

Source (SQL: Transactions)
  → Filter: WHERE year = currentYear AND status = 'completed'
  → Derived Column: 
      amountUSD = amount * exchangeRate
      quarter = CASE WHEN month <= 3 THEN 'Q1' WHEN month <= 6 THEN 'Q2' ...
  → Aggregate: 
      GROUP BY quarter, category
      SUM(amountUSD) as totalRevenue
      COUNT(*) as transactionCount
      AVG(amountUSD) as avgTransaction
  → Sink (Azure SQL: FinancialReport table)

Data Flow Transformations

Transformations phổ biến:
├── Source: SQL, Blob, Cosmos, REST API
├── Filter: WHERE clause
├── Select: Choose/rename columns
├── Derived Column: Calculated fields
├── Aggregate: GROUP BY + aggregation functions
├── Join: Inner, Left, Right, Full, Cross
├── Lookup: Enrich data từ reference table
├── Conditional Split: Route rows to different paths
├── Pivot / Unpivot: Reshape data
├── Window: Running totals, rankings
├── Flatten: Expand arrays/nested JSON
├── Surrogate Key: Auto-generate IDs
└── Sink: Write to destination

3. Parameterization & Dynamic Content

json
// Pipeline parameters
{
  "parameters": {
    "Environment": { "type": "string", "defaultValue": "dev" },
    "StartDate": { "type": "string" },
    "EndDate": { "type": "string" },
    "SourceTable": { "type": "string" }
  }
}

// Dynamic expressions
// Dataset: @dataset().tableName
// Pipeline: @pipeline().parameters.Environment

// Ví dụ Dynamic query:
"sqlReaderQuery": "@concat(
    'SELECT * FROM ', pipeline().parameters.SourceTable,
    ' WHERE ModifiedDate BETWEEN ''',
    pipeline().parameters.StartDate,
    ''' AND ''',
    pipeline().parameters.EndDate, ''''
)"

// ForEach Activity: loop qua tables
{
  "name": "ForEachTable",
  "type": "ForEach",
  "typeProperties": {
    "items": "@pipeline().parameters.TableList",
    "isSequential": false,
    "batchCount": 10,
    "activities": [
      {
        "name": "CopyTable",
        "type": "Copy",
        "typeProperties": {
          "source": {
            "type": "SqlSource",
            "sqlReaderQuery": "@concat('SELECT * FROM ', item().tableName)"
          }
        }
      }
    ]
  }
}

4. Integration Runtime

Azure IR (default):
  ✅ Managed by Microsoft, auto-scale
  ✅ Data movement between cloud sources
  ✅ Data Flow execution (Spark)
  ⚠️ Cannot access on-premises data

Self-hosted IR:
  ✅ Access on-premises SQL Server, file shares
  ✅ Access VNet-restricted resources
  ✅ Install on Windows VM (on-prem or Azure)
  ⚠️ Need machine maintenance

SSIS IR:
  ✅ Run existing SSIS packages in cloud
  ✅ Lift-and-shift SSIS workflows

5. CI/CD cho ADF

Git Integration:
├── Azure DevOps Repo (hoặc GitHub)
├── Collaboration branch: main
├── Publish branch: adf_publish (auto-generated ARM templates)

CI/CD Flow:
1. Developer: edit pipeline trong ADF Studio
2. Save: changes commit to feature branch
3. PR Review: merge to main
4. Publish: ADF generate ARM template → adf_publish branch
5. CI/CD Pipeline: deploy ARM template to staging/production

// Azure DevOps Pipeline
trigger:
  branches:
    include:
      - adf_publish

pool:
  vmImage: 'ubuntu-latest'

steps:
  - task: AzureResourceManagerTemplateDeployment@3
    inputs:
      deploymentScope: 'Resource Group'
      azureResourceManagerConnection: 'AzureServiceConnection'
      resourceGroupName: 'rg-dataplatform-prod'
      location: 'Southeast Asia'
      templateLocation: 'Linked artifact'
      csmFile: '$(System.DefaultWorkingDirectory)/adf_publish/ARMTemplateForFactory.json'
      csmParametersFile: '$(System.DefaultWorkingDirectory)/adf_publish/ARMTemplateParametersForFactory.json'
      overrideParameters: '-factoryName "adf-prod" -LS_SqlServer_connectionString "$(ProdSqlConnection)"'

6. Ví Dụ: ETL Pipeline cho Financial Reporting

Schedule: Daily at 1 AM UTC

Pipeline: DailyFinancialETL
│
├── Activity 1: LookupLastRunTime
│   → Query metadata table for last successful run
│
├── Activity 2: CopyTransactions (incremental)
│   → Source: On-prem SQL (Self-hosted IR)
│   → Sink: Azure Data Lake (Parquet format)
│   → Filter: ModifiedDate > LastRunTime
│
├── Activity 3: DataFlow_TransformFinancials
│   → Read from Data Lake
│   → Currency conversion
│   → Categorization
│   → Aggregation by region, product, period
│   → Write to Azure SQL (reporting DB)
│
├── Activity 4: StoredProcedure_UpdateSummary
│   → Refresh materialized views
│   → Update KPI tables
│
├── Activity 5: UpdateMetadata
│   → Record successful run time
│
└── On Failure: WebActivity → Send Teams notification

Tổng Kết Session 22

Bài tập: Build ADF pipeline: copy data từ Azure SQL → Data Lake (Parquet) → Data Flow transform → Azure SQL reporting table. Add incremental load pattern.