Fabric Data Factory

Fabric Data Factory

1. Data Factory in Fabric Overview

Microsoft Fabric Data Factory is the data integration service within Microsoft Fabric. It provides a unified experience for building ETL (Extract, Transform, Load) and ELT (Extract, Load, Transform) data pipelines at scale.

Data Factory vs Azure Data Factory (ADF):

Item types in Fabric Data Factory:

• Data Pipeline — Orchestration of activities (Copy, Dataflow, Notebook, Script, etc.) with dependencies, parameters, and scheduling.
• Dataflow Gen2 — Visual, low-code data transformation using Power Query Online and the M language. Outputs to lakehouse tables, warehouse, or other destinations.

Capacity-based compute:

• All pipeline and dataflow execution consumes Fabric Capacity Units (CUs).
• No need to provision separate integration runtimes.
• Fabric automatically scales compute within the capacity.
• Capacity can be paused to save costs; scheduled pipelines will queue and run when capacity resumes.

Connector ecosystem:

• 70+ built-in connectors for databases, files, SaaS services, and cloud platforms.
• Connectors are shared across Fabric workloads (pipelines, dataflows, notebooks).
• On-premises data access via the on-premises data gateway.
• Custom connectors via REST API or OData generic connectors.

Relationship to other Fabric workloads:

• Lakehouse: Primary destination for pipeline and dataflow output (Delta tables in OneLake).
• Warehouse: SQL-based destination for structured data.
• Notebook: Spark notebooks can be called as pipeline activities for complex transformations.
• KQL Database: Real-time analytics destination for streaming and event data.
• Semantic Model: Downstream consumption layer for Power BI reporting.

2. Data Pipelines

Data pipelines orchestrate data movement and transformation through a sequence of activities connected by dependencies.

Pipeline JSON structure:

{
  "name": "MyPipeline",
  "properties": {
    "activities": [
      {
        "name": "CopyFromSQL",
        "type": "Copy",
        "dependsOn": [],
        "policy": {
          "timeout": "01:00:00",
          "retry": 2,
          "retryIntervalInSeconds": 30
        },
        "typeProperties": {
          "source": { "type": "AzureSqlSource", "sqlReaderQuery": "SELECT * FROM dbo.Customers" },
          "sink": { "type": "LakehouseSink", "tableActionOption": "Overwrite" }
        }
      }
    ],
    "parameters": {
      "sourceTable": { "type": "String", "defaultValue": "dbo.Customers" }
    },
    "variables": {
      "rowCount": { "type": "String", "defaultValue": "0" }
    },
    "annotations": ["production", "daily-load"]
  }
}

Activity types:

Dependencies and execution order:

Activities use dependsOn to define execution order. Each dependency specifies one of four conditions:

"dependsOn": [
  {
    "activity": "CopyFromSQL",
    "dependencyConditions": ["Succeeded"]
  }
]

Activities with no dependsOn run in parallel at the start of the pipeline.

REST API for pipeline CRUD:

# List pipelines in a workspace
GET https://api.fabric.microsoft.com/v1/workspaces/{workspaceId}/items?type=DataPipeline

# Create a pipeline
POST https://api.fabric.microsoft.com/v1/workspaces/{workspaceId}/items
Body: { "displayName": "MyPipeline", "type": "DataPipeline", "definition": { ... } }

# Update a pipeline
PATCH https://api.fabric.microsoft.com/v1/workspaces/{workspaceId}/items/{itemId}
Body: { "displayName": "UpdatedName" }

# Delete a pipeline
DELETE https://api.fabric.microsoft.com/v1/workspaces/{workspaceId}/items/{itemId}

# Run a pipeline on demand
POST https://api.fabric.microsoft.com/v1/workspaces/{workspaceId}/items/{itemId}/jobs/instances?jobType=Pipeline

3. Copy Activity

The Copy activity moves data between 70+ supported sources and sinks with optional transformations, fault tolerance, and performance tuning.

Source connectors (selected):

Sink connectors:

Column mapping:

"translator": {
  "type": "TabularTranslator",
  "mappings": [
    {
      "source": { "name": "CustomerID", "type": "Int32" },
      "sink": { "name": "customer_id", "type": "Int64" }
    },
    {
      "source": { "name": "FullName" },
      "sink": { "name": "full_name" }
    },
    {
      "source": { "name": "SignupDate", "type": "DateTime" },
      "sink": { "name": "signup_date", "type": "DateTime", "format": "yyyy-MM-dd" }
    }
  ],
  "typeConversion": true,
  "typeConversionSettings": {
    "allowDataTruncation": false,
    "treatBooleanAsNumber": false
  }
}

If no explicit mapping is specified, columns are auto-mapped by name (case-insensitive). Unmapped source columns are ignored; unmapped sink columns receive null.

Fault tolerance:

"typeProperties": {
  "source": { ... },
  "sink": { ... },
  "enableSkipIncompatibleRow": true,
  "redirectIncompatibleRowSettings": {
    "linkedServiceName": "BlobForLogs",
    "path": "errors/incompatible-rows"
  }
}

Incompatible rows (type mismatches, constraint violations) are skipped and logged to the redirect path instead of failing the entire copy.

Staging:

"enableStaging": true,
"stagingSettings": {
  "linkedServiceName": "AzureBlobStaging",
  "path": "staging-container"
}

Staging is recommended for:

• On-premises to cloud copies (via gateway)
• Cross-region copies for better throughput
• Format conversions (e.g., Parquet to SQL)

Performance settings:

Performance tips:

• Use Parquet format for file destinations — columnar compression gives best throughput.
• Enable staging for copies involving format conversion.
• For large single-file copies, enable block-level parallelism with enablePartitionDiscovery.
• Prefer SELECT column1, column2 FROM table over SELECT * FROM table to reduce data volume.
• When copying to lakehouse, use Delta table Overwrite mode for full loads (avoids small file problem).

4. Dataflow Gen2

Dataflow Gen2 provides a visual, low-code data transformation experience powered by Power Query Online and the M language. It is the primary tool for data cleansing, shaping, and lightweight transformations in Fabric.

Dataflow Gen2 vs Dataflow Gen1:

M language basics:

M (Power Query Formula Language) is a functional language with a let/in structure:

let
    // Each step is a named expression
    Source = Sql.Database("server.database.windows.net", "AdventureWorks"),
    Navigation = Source{[Schema="Sales", Item="Customer"]}[Data],
    FilteredRows = Table.SelectRows(Navigation, each [IsActive] = true),
    SelectedColumns = Table.SelectColumns(FilteredRows, {"CustomerID", "Name", "Email", "Region"}),
    RenamedColumns = Table.RenameColumns(SelectedColumns, {
        {"CustomerID", "customer_id"},
        {"Name", "customer_name"},
        {"Email", "email"},
        {"Region", "region"}
    }),
    ChangedTypes = Table.TransformColumnTypes(RenamedColumns, {
        {"customer_id", Int64.Type},
        {"customer_name", type text},
        {"email", type text},
        {"region", type text}
    })
in
    ChangedTypes

Key M functions by category:

Merge (join) types:

Staging to lakehouse:

Dataflow Gen2 can stage intermediate data in a lakehouse for better performance:

1. Enable staging in the dataflow output settings.
2. Select the target lakehouse and table name.
3. Choose the load mode: Overwrite or Append.

Staging uses the lakehouse as a temporary buffer, enabling larger datasets to process without hitting memory limits. Data is written as Delta tables.

Performance optimization:

• Push filters early: Apply Table.SelectRows before joins and aggregations so the engine can push filters to the source (query folding).
• Limit columns early: Use Table.SelectColumns immediately after the source step to reduce data volume.
• Enable staging: Always enable lakehouse staging for dataflows processing more than a few thousand rows.
• Avoid custom functions on large datasets: Table.AddColumn with complex each expressions breaks query folding. Prefer native M functions.
• Use native query: For SQL sources, provide a native SQL query via Value.NativeQuery() when M transformations alone cannot fold.

5. Expressions & Parameters

Pipeline expressions provide dynamic content in activity properties using a rich expression language.

Expression syntax:

Expressions are enclosed in @{...} for string interpolation or used directly with @ prefix:

# String interpolation
"Hello @{pipeline().parameters.userName}, your run ID is @{pipeline().RunId}"

# Direct expression (for non-string properties)
@pipeline().parameters.maxRows

# Nested expressions
@concat('output_', formatDateTime(utcNow(), 'yyyyMMdd'), '.csv')

Pipeline system variables:

Parameters:

Parameters are defined at the pipeline level and can be set at runtime:

"parameters": {
  "sourceTable": {
    "type": "String",
    "defaultValue": "dbo.Customers"
  },
  "loadDate": {
    "type": "String",
    "defaultValue": ""
  },
  "maxRetries": {
    "type": "Int",
    "defaultValue": 3
  },
  "isFullLoad": {
    "type": "Bool",
    "defaultValue": false
  }
}

Reference parameters in expressions: @pipeline().parameters.sourceTable

Variables:

Variables are pipeline-scoped and can be set during execution:

"variables": {
  "currentStatus": { "type": "String", "defaultValue": "Running" },
  "processedCount": { "type": "String", "defaultValue": "0" },
  "fileList": { "type": "Array", "defaultValue": [] }
}

Set variables with the Set Variable activity. Append to arrays with Append Variable.

Activity output references:

Reference output from upstream activities:

# Copy activity output
@activity('CopyCustomers').output.rowsCopied
@activity('CopyCustomers').output.rowsRead
@activity('CopyCustomers').output.dataWritten
@activity('CopyCustomers').output.throughput

# Lookup activity output
@activity('GetWatermark').output.firstRow.watermarkValue
@activity('GetWatermark').output.count
@activity('GetWatermark').output.value    # Array of all rows

# Get Metadata output
@activity('CheckFile').output.itemName
@activity('CheckFile').output.exists
@activity('CheckFile').output.lastModified
@activity('CheckFile').output.childItems   # List of files in a folder

# Web activity output
@activity('CallAPI').output.statusCode
@activity('CallAPI').output.Response       # Parsed JSON response body

# Activity status
@activity('CopyCustomers').status          # Succeeded, Failed, Cancelled

Common expression functions:

6. Orchestration Patterns

Well-designed orchestration separates concerns into reusable, composable pipeline patterns.

Master-child pipeline pattern:

A master (parent) pipeline calls child pipelines using Execute Pipeline activities. This separates data movement, transformation, and reporting into independent, testable units.

{
  "name": "MasterPipeline",
  "properties": {
    "activities": [
      {
        "name": "IngestRawData",
        "type": "ExecutePipeline",
        "typeProperties": {
          "pipeline": { "referenceName": "IngestPipeline", "type": "PipelineReference" },
          "parameters": { "sourceDate": "@pipeline().parameters.loadDate" },
          "waitOnCompletion": true
        }
      },
      {
        "name": "TransformData",
        "type": "ExecutePipeline",
        "dependsOn": [{ "activity": "IngestRawData", "dependencyConditions": ["Succeeded"] }],
        "typeProperties": {
          "pipeline": { "referenceName": "TransformPipeline", "type": "PipelineReference" },
          "waitOnCompletion": true
        }
      },
      {
        "name": "RefreshReports",
        "type": "ExecutePipeline",
        "dependsOn": [{ "activity": "TransformData", "dependencyConditions": ["Succeeded"] }],
        "typeProperties": {
          "pipeline": { "referenceName": "RefreshPipeline", "type": "PipelineReference" },
          "waitOnCompletion": true
        }
      }
    ],
    "parameters": {
      "loadDate": { "type": "String", "defaultValue": "" }
    }
  }
}

ForEach with parallel execution:

Process multiple items (tables, files, partitions) in parallel:

{
  "name": "ProcessAllTables",
  "type": "ForEach",
  "typeProperties": {
    "items": {
      "value": "@pipeline().parameters.tableList",
      "type": "Expression"
    },
    "isSequential": false,
    "batchCount": 10,
    "activities": [
      {
        "name": "CopyTable",
        "type": "Copy",
        "typeProperties": {
          "source": {
            "type": "AzureSqlSource",
            "sqlReaderQuery": "SELECT * FROM @{item().schemaName}.@{item().tableName}"
          },
          "sink": {
            "type": "LakehouseSink",
            "tableActionOption": "Overwrite"
          }
        }
      }
    ]
  }
}

• isSequential: false — Items process in parallel.
• batchCount: 10 — Up to 10 items execute concurrently (max 50).
• @item() — References the current item in the ForEach loop.

Error handling with If Condition:

Branch pipeline execution based on the success or failure of upstream activities:

{
  "name": "HandleCopyResult",
  "type": "IfCondition",
  "dependsOn": [{ "activity": "CopyData", "dependencyConditions": ["Completed"] }],
  "typeProperties": {
    "expression": {
      "value": "@equals(activity('CopyData').status, 'Succeeded')",
      "type": "Expression"
    },
    "ifTrueActivities": [
      {
        "name": "LogSuccess",
        "type": "Script",
        "typeProperties": {
          "scripts": [{ "text": "INSERT INTO dbo.PipelineLog VALUES ('@{pipeline().RunId}', 'Succeeded', GETUTCDATE())" }]
        }
      }
    ],
    "ifFalseActivities": [
      {
        "name": "LogFailure",
        "type": "Script",
        "typeProperties": {
          "scripts": [{ "text": "INSERT INTO dbo.PipelineLog VALUES ('@{pipeline().RunId}', 'Failed', GETUTCDATE())" }]
        }
      },
      {
        "name": "SendAlert",
        "type": "Web",
        "typeProperties": {
          "url": "https://webhook.site/notify",
          "method": "POST",
          "body": { "pipeline": "@{pipeline().Pipeline}", "error": "@{activity('CopyData').error.message}" }
        }
      }
    ]
  }
}

Watermark pattern for incremental loads:

Load only new or changed data by tracking a high-watermark value:

1. Lookup — Read the current watermark from a control table.
2. Copy — Copy rows where the change column > watermark.
3. Script — Update the control table with the new watermark value.

Lookup(GetWatermark) → Copy(IncrementalCopy) → Script(UpdateWatermark)

The watermark column is typically ModifiedDate, RowVersion, or an auto-increment ID.

7. Scheduling & Triggers

Pipelines can be triggered on a schedule, by events, or manually.

Schedule triggers:

Configure recurring execution using frequency and interval:

{
  "type": "ScheduleTrigger",
  "typeProperties": {
    "recurrence": {
      "frequency": "Day",
      "interval": 1,
      "startTime": "2025-01-01T06:00:00Z",
      "timeZone": "UTC",
      "schedule": {
        "hours": [6],
        "minutes": [0]
      }
    }
  }
}

Recurrence examples:

Tumbling window triggers:

Process data in fixed, non-overlapping time windows. Useful for time-partitioned loads:

• Each window has a defined start and end time.
• Missed windows are automatically backfilled.
• Dependencies can be set between tumbling windows of different pipelines.
• Retry policies apply per window.

Event-based triggers:

Trigger pipeline execution when specific events occur:

• File arrival in Blob Storage or ADLS Gen2 (e.g., new file in input/ folder).
• Custom events via Event Grid.
• Storage event triggers monitor a specific container and path pattern.

On-demand execution:

Run a pipeline manually via the Fabric UI or REST API:

POST https://api.fabric.microsoft.com/v1/workspaces/{workspaceId}/items/{pipelineId}/jobs/instances?jobType=Pipeline
Body: { "executionData": { "parameters": { "loadDate": "2025-01-15" } } }

Parameters at trigger time:

Schedule triggers can pass parameters to the pipeline at execution time:

"pipelines": [{
  "pipelineReference": { "referenceName": "DailyLoad" },
  "parameters": {
    "loadDate": "@trigger().scheduledTime",
    "windowStart": "@trigger().outputs.windowStartTime",
    "windowEnd": "@trigger().outputs.windowEndTime"
  }
}]

8. Connectors & Connections

Connections define how Fabric connects to external data sources. They are workspace-level resources shared across pipelines and dataflows.

Connection types:

Authentication methods:

Connection best practices:

• Prefer managed identity for Azure-to-Azure connectivity — no credential rotation needed.
• Use service principal for CI/CD pipelines that deploy and run pipelines programmatically.
• Store connection strings and secrets in Azure Key Vault; reference them via linked Key Vault connections.
• Use workspace-level connections to avoid duplicate credentials across pipelines.
• Rotate shared keys and client secrets on a regular schedule (90 days recommended).

On-premises data gateway:

For accessing data behind firewalls (SQL Server, Oracle, file shares):

1. Install the on-premises data gateway on a machine with network access to the data source.
2. Register the gateway in the Fabric admin portal.
3. Create a connection that references the gateway.
4. Use the connection in pipelines and dataflows.

Gateway considerations:

• Deploy in a cluster (2+ nodes) for high availability.
• Monitor gateway health via the Fabric admin portal.
• Keep the gateway software updated (auto-update recommended).
• Place the gateway machine close to the data source to minimize latency.

9. Error Handling & Retry

Production pipelines must handle errors gracefully to avoid data loss and provide visibility into failures.

Activity-level retry policies:

"policy": {
  "timeout": "01:00:00",
  "retry": 3,
  "retryIntervalInSeconds": 30,
  "secureOutput": false,
  "secureInput": false
}

Try-catch pattern:

Use dependency conditions to create try-catch-like behavior:

CopyData (main activity)
  ├── [Succeeded] → LogSuccess
  └── [Failed] → HandleError
                    ├── LogError (Script activity)
                    └── SendNotification (Web activity)

The key is using dependencyConditions: ["Failed"] to route to error-handling activities.

Logging errors to a table:

{
  "name": "LogError",
  "type": "Script",
  "dependsOn": [{ "activity": "CopyData", "dependencyConditions": ["Failed"] }],
  "typeProperties": {
    "scripts": [{
      "text": "INSERT INTO dbo.ErrorLog (RunId, PipelineName, ActivityName, ErrorMessage, ErrorTime) VALUES ('@{pipeline().RunId}', '@{pipeline().Pipeline}', 'CopyData', '@{activity('CopyData').error.message}', GETUTCDATE())"
    }]
  }
}

Pipeline failure notifications:

Use a Web activity to call a webhook (Teams, Slack, email service) on failure:

{
  "name": "NotifyOnFailure",
  "type": "Web",
  "dependsOn": [{ "activity": "CopyData", "dependencyConditions": ["Failed"] }],
  "typeProperties": {
    "url": "https://outlook.office.com/webhook/...",
    "method": "POST",
    "headers": { "Content-Type": "application/json" },
    "body": {
      "title": "Pipeline Failed: @{pipeline().Pipeline}",
      "text": "Run @{pipeline().RunId} failed at @{utcNow()}. Error: @{activity('CopyData').error.message}"
    }
  }
}

Best practices:

• Always set retry on activities that call external services (Copy, Web, Script) — transient errors are common.
• Set explicit timeout values — the default 7-day timeout is too long for production pipelines.
• Log all errors to a persistent store (lakehouse table or SQL table) for trend analysis.
• Use secureInput: true on activities that handle credentials (e.g., Web activity calling a token endpoint).
• Implement a "dead letter" pattern: move failed records to a separate table for manual review instead of blocking the entire pipeline.

10. Monitoring

Monitoring pipeline and dataflow execution is essential for maintaining data freshness and reliability.

Fabric Monitoring Hub:

The Monitoring Hub provides a centralized view of all Fabric item executions:

• Filter by workspace, item type (pipeline, dataflow, notebook), status, and time range.
• View pipeline run history with status, duration, start time, and trigger type.
• Drill into individual runs to see activity-level details.
• View data read/written, throughput, and duration per activity.

Pipeline run states:

REST API for monitoring:

# List pipeline runs
GET https://api.fabric.microsoft.com/v1/workspaces/{workspaceId}/items/{pipelineId}/jobs/instances

# Get specific run details
GET https://api.fabric.microsoft.com/v1/workspaces/{workspaceId}/items/{pipelineId}/jobs/instances/{runId}

# Cancel a running pipeline
POST https://api.fabric.microsoft.com/v1/workspaces/{workspaceId}/items/{pipelineId}/jobs/instances/{runId}/cancel

Diagnostic logs:

Enable diagnostic settings to send pipeline logs to:

• Log Analytics workspace (for KQL queries and dashboards)
• Storage account (for long-term retention)
• Event Hub (for real-time alerting)

Alerting on failure:

Configure alerts via:

1. Fabric workspace settings — Built-in email notifications for pipeline failures.
2. Azure Monitor alerts — KQL-based alerts on diagnostic logs.
3. Custom webhook — Web activity in the pipeline that fires on failure paths.
4. Teams / Slack — Incoming webhook to a Teams channel for real-time notifications.

Performance baselines:

Track and baseline pipeline performance:

• Monitor run duration trends — increasing duration signals schema drift, data growth, or resource contention.
• Track rows read vs rows written — discrepancies indicate data quality issues or filter drift.
• Monitor capacity utilization — high CU consumption may require capacity scaling.
• Set up SLA alerts — notify if a pipeline does not complete within its expected window.

11. Migration from ADF

Teams moving from Azure Data Factory to Fabric Data Factory should understand the key differences and migration path.

Key differences:

Migration considerations:

• No self-hosted IR in Fabric: Use the on-premises data gateway instead for on-premises data access.
• Linked services become connections: Recreate each linked service as a Fabric workspace connection.
• Mapping Data Flows become Dataflow Gen2 or Notebooks: Mapping Data Flow logic needs to be rewritten as Power Query M (Dataflow Gen2) or PySpark (Notebook activities).
• Trigger migration: Schedule and tumbling window triggers have direct equivalents. Custom event triggers use the same Event Grid integration.
• Pipeline JSON compatibility: Pipeline activity definitions are largely compatible, but connection references need updating.
• Global parameters: ADF global parameters map to Fabric workspace variables or pipeline parameters.

Feature parity gaps (as of 2025):

• No SSIS Integration Runtime in Fabric — SSIS packages must be re-implemented as pipelines or notebooks.
• Limited custom activity support — ADF's Azure Batch custom activities are not directly available.
• No private endpoint support for Fabric Data Factory connections (use gateway instead).
• Mapping Data Flow debug sessions are not available in Fabric; use Dataflow Gen2 preview or notebook debugging.

12. Common Patterns

Pattern 1: Full and Incremental Load from SQL to Lakehouse

Load a SQL table to a lakehouse using a full load on first run and incremental loads thereafter.

{
  "name": "SQLToLakehouse_Incremental",
  "properties": {
    "activities": [
      {
        "name": "GetWatermark",
        "type": "Lookup",
        "typeProperties": {
          "source": {
            "type": "LakehouseSource",
            "sqlReaderQuery": "SELECT COALESCE(MAX(watermark_value), '1900-01-01') AS watermark FROM control.watermark_table WHERE table_name = '@{pipeline().parameters.tableName}'"
          },
          "firstRowOnly": true
        }
      },
      {
        "name": "IncrementalCopy",
        "type": "Copy",
        "dependsOn": [{ "activity": "GetWatermark", "dependencyConditions": ["Succeeded"] }],
        "typeProperties": {
          "source": {
            "type": "AzureSqlSource",
            "sqlReaderQuery": "SELECT * FROM @{pipeline().parameters.tableName} WHERE ModifiedDate > '@{activity('GetWatermark').output.firstRow.watermark}' AND ModifiedDate <= '@{pipeline().parameters.loadDate}'"
          },
          "sink": {
            "type": "LakehouseSink",
            "tableActionOption": "Append"
          }
        }
      },
      {
        "name": "UpdateWatermark",
        "type": "Script",
        "dependsOn": [{ "activity": "IncrementalCopy", "dependencyConditions": ["Succeeded"] }],
        "typeProperties": {
          "scripts": [{
            "text": "MERGE control.watermark_table AS t USING (SELECT '@{pipeline().parameters.tableName}' AS table_name, '@{pipeline().parameters.loadDate}' AS watermark_value) AS s ON t.table_name = s.table_name WHEN MATCHED THEN UPDATE SET watermark_value = s.watermark_value WHEN NOT MATCHED THEN INSERT (table_name, watermark_value) VALUES (s.table_name, s.watermark_value);"
          }]
        }
      }
    ],
    "parameters": {
      "tableName": { "type": "String" },
      "loadDate": { "type": "String", "defaultValue": "@utcNow()" }
    }
  }
}

Pattern 2: Dataflow Gen2 Data Cleaning Pipeline

Ingest raw data, clean it with a Dataflow Gen2, and validate output quality.

{
  "name": "DataCleaningPipeline",
  "properties": {
    "activities": [
      {
        "name": "CopyRawData",
        "type": "Copy",
        "typeProperties": {
          "source": { "type": "AzureBlobSource", "recursive": true },
          "sink": { "type": "LakehouseSink", "tableActionOption": "Overwrite" }
        }
      },
      {
        "name": "CleanWithDataflow",
        "type": "DataflowActivity",
        "dependsOn": [{ "activity": "CopyRawData", "dependencyConditions": ["Succeeded"] }],
        "typeProperties": {
          "dataflowReference": {
            "referenceName": "CleanCustomerDataflow",
            "type": "DataflowReference"
          }
        }
      },
      {
        "name": "ValidateRowCount",
        "type": "Lookup",
        "dependsOn": [{ "activity": "CleanWithDataflow", "dependencyConditions": ["Succeeded"] }],
        "typeProperties": {
          "source": {
            "type": "LakehouseSource",
            "sqlReaderQuery": "SELECT COUNT(*) AS row_count FROM cleaned_customers"
          },
          "firstRowOnly": true
        }
      },
      {
        "name": "CheckQuality",
        "type": "IfCondition",
        "dependsOn": [{ "activity": "ValidateRowCount", "dependencyConditions": ["Succeeded"] }],
        "typeProperties": {
          "expression": {
            "value": "@greater(activity('ValidateRowCount').output.firstRow.row_count, 0)",
            "type": "Expression"
          },
          "ifTrueActivities": [
            {
              "name": "LogSuccess",
              "type": "Script",
              "typeProperties": {
                "scripts": [{ "text": "INSERT INTO control.pipeline_log VALUES ('@{pipeline().RunId}', 'DataCleaning', 'Succeeded', @{activity('ValidateRowCount').output.firstRow.row_count}, GETUTCDATE())" }]
              }
            }
          ],
          "ifFalseActivities": [
            {
              "name": "FailPipeline",
              "type": "Fail",
              "typeProperties": {
                "message": "Data quality check failed: cleaned table has 0 rows",
                "errorCode": "DQ_ZERO_ROWS"
              }
            }
          ]
        }
      }
    ]
  }
}

Companion Dataflow Gen2 M query (CleanCustomerDataflow):

let
    Source = Lakehouse.Contents(null){[workspaceId="<workspace-id>"]}[Data]{[lakehouseId="<lakehouse-id>"]}[Data],
    RawCustomers = Source{[Id="raw_customers", ItemKind="Table"]}[Data],
    RemoveNulls = Table.SelectRows(RawCustomers, each [Email] <> null and [Name] <> null),
    TrimWhitespace = Table.TransformColumns(RemoveNulls, {
        {"Name", Text.Trim, type text},
        {"Email", Text.Trim, type text}
    }),

…(truncated)