Data Modelling for Investment, Risk, and Performance Analytics on the Fabric Medallion Architecture

AI-generated content. Platform facts cite sources [Sn]; the data model, code, and walkthrough are original AI-authored examples, not yet human-verified.

Overview#

Investment operations, risk, and performance teams share one hard problem: the same underlying facts (positions, transactions, returns, risk measures) need to reach a warehouse-grade, auditable, BI-ready form, while also staying traceable back to a system of record and to whatever market or risk-engine data was blended in along the way. This walkthrough builds a simplified but complete, working prototype of that pipeline on Microsoft Fabric, using the medallion pattern: bronze and silver share a single lakehouse, gold is a dbt-built Fabric Warehouse, orchestration is metadata-driven, and Power BI consumes gold through Direct Lake ^{[S1] [S2]}.

The three ingestion paths are deliberately different in kind, because the three source types are different in kind:

• Files (extracts, custodian statements, benchmark files) land through a Data Factory Copy activity, because Copy activity gives full manual control over parallelism and format handling for arbitrary file-shaped sources ^[S3].
• The portfolio/transactions system of record — an operational relational database — is mirrored into OneLake as Delta with no pipeline at all, because Fabric Mirroring is purpose-built for exactly this: a fully managed, serverless replication service that continuously converts an operational database into Delta Parquet in OneLake without hand-built ETL ^[S4].
• Market data and risk-engine outputs arrive by calling REST APIs on a schedule, because that data doesn't live in a database Fabric can mirror — it has to be pulled ^[S3].

All three paths are driven by control tables in a Fabric SQL database, sqldb_meta, so adding a new file feed, a new mirrored table, or a new API source is a metadata change, not a pipeline rewrite. Bronze and silver both live in one lakehouse, lh_analytics, per this prototype's simplified design — bronze is raw and append-only, silver is PySpark-conformed and deduplicated, and the mirrored source surfaces into the same lakehouse via a OneLake shortcut so it feeds the same silver step as everything else ^{[S5] [S6]}. Gold lives in a Fabric Warehouse, wh_gold, built entirely with dbt against the dbt-fabric adapter, scheduled by an Apache Airflow job inside Data Factory ^{[S7] [S3]}. Power BI reads gold through Direct Lake. An Azure Function posts granular pipeline telemetry to an Eventhouse, eh_ops, for KQL-based operational monitoring, while sqldb_meta keeps the coarser run-level audit trail ^{[S8] [S5]}.

1. Setup#

To follow along, provision these Fabric items in one workspace (a trial or Fabric capacity works fine for a prototype):

You'll also need: a Data Factory pipeline canvas in the workspace for Copy activities and the API-pull pipeline; the dbt-fabric adapter installed locally or in a CI runner (pip install dbt-fabric) to build gold ^[S7]; an Apache Airflow environment inside Data Factory to schedule dbt build ^{[S3] [S7]}; and an Azure Function (Python) with network access to the workspace for telemetry posting.

Every Fabric SQL database automatically gets a read-only SQL analytics endpoint and continuously, near-real-time replicates its data to OneLake as Parquet, so sqldb_meta itself is queryable analytically with zero extra setup — useful later when you want to report on run history from Power BI as well as T-SQL ^[S5].

2. Ingest#

2.1 Control tables in sqldb_meta#

Everything downstream is driven off three tables: what to ingest (etl_entity), what happened (etl_run), and where each source left off (etl_watermark).

-- sqldb_meta: control-table DDL

CREATE TABLE dbo.etl_entity (
    entity_id        INT IDENTITY(1,1) PRIMARY KEY,
    entity_name      NVARCHAR(128)   NOT NULL,       -- e.g. 'positions_file', 'portfolio_mirror', 'market_data_api'
    source_kind      NVARCHAR(32)    NOT NULL,        -- 'FILE' | 'MIRROR' | 'API'
    source_path      NVARCHAR(512)   NULL,            -- file container/path, API base URL, or mirrored DB name
    target_lakehouse NVARCHAR(128)   NOT NULL DEFAULT 'lh_analytics',
    target_table     NVARCHAR(128)   NOT NULL,        -- bronze table name
    schedule_cron    NVARCHAR(64)    NULL,            -- for API/file pulls; NULL for mirrored entities
    is_active        BIT             NOT NULL DEFAULT 1,
    created_at        DATETIME2       NOT NULL DEFAULT SYSUTCDATETIME()
);

CREATE TABLE dbo.etl_run (
    run_id           BIGINT IDENTITY(1,1) PRIMARY KEY,
    entity_id        INT             NOT NULL REFERENCES dbo.etl_entity(entity_id),
    pipeline_name    NVARCHAR(128)   NOT NULL,
    started_at       DATETIME2       NOT NULL,
    finished_at      DATETIME2       NULL,
    status           NVARCHAR(16)    NOT NULL DEFAULT 'RUNNING',  -- RUNNING | SUCCEEDED | FAILED
    rows_read        BIGINT          NULL,
    rows_written     BIGINT          NULL,
    error_message    NVARCHAR(MAX)   NULL
);

CREATE TABLE dbo.etl_watermark (
    entity_id        INT             NOT NULL PRIMARY KEY REFERENCES dbo.etl_entity(entity_id),
    watermark_value  NVARCHAR(128)   NOT NULL,    -- last-seen timestamp, file name, or API cursor, as text
    watermark_type   NVARCHAR(32)    NOT NULL,    -- 'DATETIME' | 'FILENAME' | 'CURSOR'
    updated_at       DATETIME2       NOT NULL DEFAULT SYSUTCDATETIME()
);

Each Data Factory pipeline run does the same three things against these tables: read its etl_entity row and current etl_watermark at the start, insert an etl_run row when it starts, and update both the watermark and the run row (rows read/written, status, timestamps) when it finishes. sqldb_meta is a Fabric SQL database, which shares the transactional Azure SQL engine, replicates automatically to OneLake, and enforces Microsoft Entra ID-only authentication — a sensible home for a control plane that other pipelines and Power BI reports will also want to query ^[S5].

2.2 Files via Data Factory Copy activity#

For file-shaped sources (custodian position extracts, benchmark constituent files, static reference data), a Copy activity is the right tool because it gives full, explicit control over parallelism, schema mapping, and file-format handling — as opposed to Copy job's simplified, opinionated defaults ^[S3]. The pipeline is parameterized entirely from the etl_entity row so one pipeline definition serves every file source.

{
  "pipeline": "pl_copy_file_source",
  "parameters": {
    "entityId": 101,
    "sourceContainer": "raw-custodian-a",
    "sourcePathPattern": "positions/{yyyy}/{MM}/{dd}/*.csv",
    "targetLakehouse": "lh_analytics",
    "targetBronzeTable": "bronze_custodian_a_positions",
    "watermarkColumn": "file_modified_utc",
    "fileFormat": "DelimitedText",
    "columnDelimiter": ",",
    "firstRowAsHeader": true
  }
}

At runtime, a Lookup activity reads the current watermark for entityId from sqldb_meta.dbo.etl_watermark, a filter/GetMetadata pair selects only files newer than that watermark, the Copy activity lands matching files into the bronze area of lh_analytics, and a final Stored Procedure/Script activity writes the new watermark and closes out the etl_run row. Because the whole pattern is parameter-driven, onboarding a fourth or fifth file feed is an INSERT into etl_entity, not a new pipeline.

2.3 Mirroring the system of record#

The portfolio and transactions system of record is an operational relational database — exactly the shape Fabric Mirroring exists for. Mirroring is a fully managed, serverless replication service: point it at the source database, and it continuously converts the data to Parquet files in Delta Lake format inside OneLake, with no ETL pipeline to build or maintain ^[S4]. For an Azure SQL Database source specifically, Mirroring is documented as a low-latency, low-cost path to replicate an existing operational estate into OneLake without touching the source ^[S5]. Under normal conditions, changes can propagate from source to OneLake in as little as 15 seconds, because an internal replicator polls a landing zone at high frequency and merges incremental Delta files as they arrive ^[S4].

This prototype treats the mirrored database as outside the bronze/silver boundary of lh_analytics in physical terms — Mirroring writes its own Delta tables in its own OneLake location — but logically joins it into the same lakehouse via a OneLake shortcut: a lakehouse Tables-folder shortcut pointed at the mirrored database's Delta tables. Because a shortcut target holding Delta Parquet data is automatically synced and registered as a table, the mirrored portfolio and transaction tables show up in lh_analytics as if they were native lakehouse tables, with no copy and no pipeline ^{[S6] [S1]}. Silver-layer PySpark code can then read lh_analytics.mirror_portfolio_shortcut.transactions exactly like any other bronze table.

No control-table row of type MIRROR triggers a Data Factory pipeline run — mirroring is continuous and pipeline-free by design — but this prototype still records a MIRROR row in etl_entity so the same etl_run/lineage story covers it (with schedule_cron left NULL and the "run" representing the periodic silver-conformance job that reads the shortcut, not a copy operation).

2.4 Scheduled API pulls for market data and risk-engine output#

Market data (prices, FX rates, curve points) and risk-engine outputs (VaR, sensitivities, scenario results) typically live behind REST APIs, not in a database Fabric can mirror, so these are pulled by scheduled Data Factory pipelines calling a Web/REST activity ^[S3]. The parameter shape mirrors the file pattern, with the addition of pagination and an incremental window:

{
  "pipeline": "pl_api_pull",
  "parameters": {
    "entityId": 201,
    "apiBaseUrl": "https://market-data.internal/api/v1/eod-prices",
    "authMode": "OAuthClientCredentials",
    "incrementalWindow": {
      "fromWatermark": true,
      "windowField": "asOfDate",
      "lookbackDays": 2
    },
    "pagination": {
      "style": "cursor",
      "pageSizeParam": "pageSize",
      "pageSize": 5000,
      "cursorParam": "continuationToken"
    },
    "targetLakehouse": "lh_analytics",
    "targetBronzeTable": "bronze_market_eod_prices"
  }
}

The pipeline loops a Web activity over pages using the continuationToken returned by each response, appends each page's JSON to a staging file, and hands the batch to a Copy or notebook activity that lands it in bronze. The lookbackDays: 2 window is deliberate for market/risk feeds: it re-pulls a couple of days behind the watermark on every run to absorb source-side restatements (a risk engine correcting yesterday's VaR after a late trade booking, for example) — silver-layer dedupe on business key plus _ingested_at ordering (section 4) resolves which version wins, rather than assuming the source never republishes a prior asOfDate.

3. Bronze in lh_analytics#

Bronze is intentionally dumb: append-only, source-shaped, one folder/schema per source, with three columns every bronze table carries regardless of source so silver and audit code can treat all bronze tables uniformly:

• _ingested_at — UTC timestamp the row was written to bronze.
• _source — the entity_name from sqldb_meta.dbo.etl_entity that produced the row.
• _watermark — the watermark value in effect for this batch (file timestamp, API cursor, or mirror shortcut read time).

# PySpark: append a raw batch into bronze, tagged with lineage columns.
from pyspark.sql import functions as F

raw_df = (
    spark.read
    .option("header", True)
    .option("inferSchema", True)
    .csv("Files/landing/custodian_a/positions/2026/07/06/*.csv")
)

bronze_df = (
    raw_df
    .withColumn("_ingested_at", F.current_timestamp())
    .withColumn("_source", F.lit("positions_file"))
    .withColumn("_watermark", F.lit("2026-07-06T00:00:00Z"))
)

(
    bronze_df.write
    .format("delta")
    .mode("append")
    .saveAsTable("bronze_custodian_a_positions")
)

Bronze tables are never overwritten or deduplicated — that discipline is what makes bronze replayable if a silver transformation bug is discovered later; the fix is to reprocess bronze into a corrected silver, not to re-pull the source ^{[S1] [S2]}. Mirrored data does not get its own bronze copy in this design — because the mirror shortcut already exposes it as Delta in lh_analytics, the append-only "bronze" role for that source is effectively played by the mirrored Delta table's own history, and silver reads directly from the shortcut.

4. Silver in lh_analytics#

Silver applies exactly the work bronze deliberately skips: schema enforcement, validation, and deduplication on a business key, producing the six conformed entities the rest of the pipeline depends on — positions, transactions, instruments, portfolios, returns_daily, risk_measures.

4.1 Conforming positions#

from pyspark.sql import functions as F
from pyspark.sql.window import Window

bronze_positions = spark.read.table("bronze_custodian_a_positions")

# Schema enforcement: cast to expected types, drop rows that fail to parse.
typed = (
    bronze_positions
    .withColumn("position_date", F.to_date("PositionDate"))
    .withColumn("quantity", F.col("Quantity").cast("decimal(18,6)"))
    .withColumn("market_value", F.col("MarketValue").cast("decimal(18,2)"))
    .withColumn("portfolio_id", F.col("PortfolioId").cast("string"))
    .withColumn("instrument_id", F.col("InstrumentId").cast("string"))
    .filter(
        F.col("position_date").isNotNull()
        & F.col("portfolio_id").isNotNull()
        & F.col("instrument_id").isNotNull()
    )
)

# Dedupe on business key: keep the latest _ingested_at per (portfolio, instrument, date).
business_key = ["portfolio_id", "instrument_id", "position_date"]
window = Window.partitionBy(*business_key).orderBy(F.col("_ingested_at").desc())

silver_positions = (
    typed
    .withColumn("_rn", F.row_number().over(window))
    .filter("_rn = 1")
    .drop("_rn")
)

(
    silver_positions.write
    .format("delta")
    .mode("overwrite")
    .option("overwriteSchema", "true")
    .saveAsTable("silver_positions")
)

4.2 Conforming risk_measures — an as-of, point-in-time join#

Risk measures need one extra piece of reasoning that positions don't: a risk figure (VaR, delta, duration) is only meaningful as of the portfolio composition it was computed against, and the risk-engine feed and the portfolio/position feed do not necessarily land at the same instant. Naively joining "today's risk output" to "today's positions" on date alone silently mismatches a risk figure against a position snapshot it was never computed from if either feed is late. The safer join is an as-of join keyed on the risk engine's own stated as_of_date, not the ingestion date:

risk_bronze = spark.read.table("bronze_risk_engine_output")
portfolio_silver = spark.read.table("silver_portfolios")

typed_risk = (
    risk_bronze
    .withColumn("as_of_date", F.to_date("AsOfDate"))
    .withColumn("portfolio_id", F.col("PortfolioId").cast("string"))
    .withColumn("risk_metric", F.col("RiskMetric").cast("string"))     # e.g. 'VAR_95_1D', 'DURATION'
    .withColumn("risk_value", F.col("RiskValue").cast("decimal(18,6)"))
    .filter(F.col("as_of_date").isNotNull() & F.col("portfolio_id").isNotNull())
)

# As-of join: attach the portfolio dimension snapshot valid on that risk measure's as_of_date,
# not the ingestion date — point-in-time correctness for risk depends on which portfolio
# composition the risk engine actually consumed, which can lag or lead the ingest clock.
enriched_risk = typed_risk.join(
    portfolio_silver.select("portfolio_id", "portfolio_name", "base_currency"),
    on="portfolio_id",
    how="left",
)

business_key = ["portfolio_id", "risk_metric", "as_of_date"]
window = Window.partitionBy(*business_key).orderBy(F.col("_ingested_at").desc())

silver_risk_measures = (
    enriched_risk
    .withColumn("_rn", F.row_number().over(window))
    .filter("_rn = 1")
    .drop("_rn")
)

(
    silver_risk_measures.write
    .format("delta")
    .mode("overwrite")
    .option("overwriteSchema", "true")
    .saveAsTable("silver_risk_measures")
)

Inference: the as-of join reasoning above (keying deduplication and enrichment off the risk engine's own as_of_date rather than ingestion time) is original prototype design, not a Fabric platform behavior — it is standard risk-analytics practice applied to this pipeline, not something any cited source asserts. The Delta table mechanics it runs on (schema-on-write, transaction-log versioning, overwrite semantics) are grounded ^{[S1] [S2]}.

silver_transactions, silver_instruments, silver_portfolios, and silver_returns_daily follow the same shape as silver_positions — type casting, not-null filtering on business keys, and a row_number() window dedupe — and are omitted here for brevity but live alongside these two in lh_analytics.

5. Gold with dbt, targeting wh_gold#

Gold is built entirely by dbt against the dbt-fabric adapter, which targets a Fabric Warehouse as a first-class dbt destination — installed with pip install dbt-fabric and pointed at wh_gold's SQL analytics endpoint through profiles.yml ^[S7].

5.1 sources.yml — cross-database sources into lh_analytics silver#

dbt's source() function models the silver tables as external inputs using Fabric's three-part naming, so a single T-SQL statement in a staging view can read across the lakehouse's SQL analytics endpoint into the warehouse project boundary with zero data movement ^{[S6] [S5]}.

# models/staging/sources.yml
version: 2

sources:
  - name: lh_silver
    database: lh_analytics
    schema: dbo
    tables:
      - name: positions
        identifier: silver_positions
      - name: transactions
        identifier: silver_transactions
      - name: instruments
        identifier: silver_instruments
      - name: portfolios
        identifier: silver_portfolios
      - name: returns_daily
        identifier: silver_returns_daily
      - name: risk_measures
        identifier: silver_risk_measures

5.2 Staging views#

-- models/staging/stg_positions.sql
{{ config(materialized='view') }}

select
    portfolio_id,
    instrument_id,
    position_date,
    quantity,
    market_value,
    _source,
    _ingested_at
from {{ source('lh_silver', 'positions') }}

-- models/staging/stg_returns.sql
{{ config(materialized='view') }}

select
    portfolio_id,
    return_date,
    benchmark_id,
    portfolio_return,
    benchmark_return,
    _source,
    _ingested_at
from {{ source('lh_silver', 'returns_daily') }}

Staging models are views, not tables: a view over the cross-database source keeps them cheap to rebuild on every dbt build and pushes no storage duplication into wh_gold for a layer that does no real transformation.

5.3 Dimensions#

-- models/marts/dim_date.sql
{{ config(materialized='table') }}

with date_spine as (
    {{ dbt_utils.date_spine(
        datepart="day",
        start_date="cast('2015-01-01' as date)",
        end_date="cast('2035-01-01' as date)"
    ) }}
)
select
    cast(format(date_day, 'yyyyMMdd') as int) as date_key,
    date_day as calendar_date,
    year(date_day) as calendar_year,
    datepart(quarter, date_day) as calendar_quarter,
    month(date_day) as calendar_month,
    datepart(iso_week, date_day) as iso_week
from date_spine

-- models/marts/dim_instrument.sql
{{ config(materialized='table') }}

-- SCD Type 2: instrument attributes (rating, sector, issuer) do change over time, and
-- historical facts (fact_positions, fact_risk_measures) must join to the attribute values
-- that were true as of the fact's date, not today's values. A plain overwrite (SCD Type 1)
-- would silently rewrite history every time an instrument is reclassified.
select
    instrument_id,
    instrument_name,
    asset_class,
    sector,
    issuer_rating,
    effective_from,
    effective_to,
    is_current
from {{ ref('stg_instrument_history') }}

-- models/marts/dim_portfolio.sql
{{ config(materialized='table') }}

select distinct
    portfolio_id,
    portfolio_name,
    base_currency,
    mandate_type
from {{ ref('stg_portfolios') }}

-- models/marts/dim_benchmark.sql
{{ config(materialized='table') }}

select distinct
    benchmark_id,
    benchmark_name,
    benchmark_currency
from {{ ref('stg_returns') }}

dim_instrument is the one dimension modeled as SCD Type 2 in this design, because instrument classification (sector, issuer rating) changes over the life of an instrument and risk/performance facts need to join to the classification that was in force on the fact's own date — the other three dimensions (dim_date, dim_portfolio, dim_benchmark) change rarely enough, or are looked up by natural business processes closely enough to "now", that this prototype treats them as simple overwrite dimensions. Inference: this SCD choice is a data-modeling decision, not a Fabric or dbt platform behavior.

5.4 Incremental fact models — and the dbt-fabric materialization caveat#

-- models/marts/fact_positions.sql
{{ config(
    materialized='incremental',
    unique_key=['portfolio_id', 'instrument_id', 'position_date']
) }}

select
    p.portfolio_id,
    p.instrument_id,
    cast(format(p.position_date, 'yyyyMMdd') as int) as date_key,
    p.quantity,
    p.market_value
from {{ ref('stg_positions') }} p

{% if is_incremental() %}
where p.position_date >= dateadd(day, -5, (select max(convert(date, cast(date_key as varchar(8)), 112)) from {{ this }}))
{% endif %}

-- models/marts/fact_returns_daily.sql
{{ config(
    materialized='incremental',
    unique_key=['portfolio_id', 'return_date', 'benchmark_id']
) }}

select
    r.portfolio_id,
    r.benchmark_id,
    cast(format(r.return_date, 'yyyyMMdd') as int) as date_key,
    r.portfolio_return,
    r.benchmark_return,
    r.portfolio_return - r.benchmark_return as active_return
from {{ ref('stg_returns') }} r

{% if is_incremental() %}
where r.return_date >= dateadd(day, -5, (select max(convert(date, cast(date_key as varchar(8)), 112)) from {{ this }}))
{% endif %}

Both fact models filter to a rolling window (dateadd(day, -5, ...)) rather than only strictly-new rows, to reabsorb the same late-arriving corrections discussed in section 2.4.

The incremental materialization name is honest but has a real caveat on this adapter that's worth stating plainly rather than assuming Databricks/Snowflake-style incremental semantics: because Fabric Warehouse's T-SQL surface doesn't support the full command set (MERGE, ALTER TABLE ADD/ALTER/DROP COLUMN, TRUNCATE) that dbt's incremental strategies normally rely on, the dbt-fabric adapter implements these operations by translating them into CREATE TABLE AS SELECT (CTAS) plus DROP/CREATE sequences under the hood ^[S7]. In practice that means an "incremental" run on this adapter can still involve materializing a full replacement table rather than a true row-level merge into the existing one — so the date-window filter here controls how much new source data is scanned and joined, but it does not guarantee the underlying warehouse operation only touches the changed rows physically. Teams should benchmark dbt run duration on this adapter directly rather than assuming incremental cost scales the way it would on a MERGE-native platform. Inference: the practical consequence for this specific project's table sizes is not measured here — the CTAS/DROP-CREATE mechanism itself is grounded ^[S7], the sizing implication is reasoning applied to it.

-- models/marts/fact_risk_measures.sql
{{ config(
    materialized='incremental',
    unique_key=['portfolio_id', 'risk_metric', 'as_of_date']
) }}

select
    portfolio_id,
    risk_metric,
    cast(format(as_of_date, 'yyyyMMdd') as int) as date_key,
    risk_value
from {{ ref('stg_risk_measures') }}

{% if is_incremental() %}
where as_of_date >= dateadd(day, -5, (select max(convert(date, cast(date_key as varchar(8)), 112)) from {{ this }}))
{% endif %}

-- models/marts/fact_transactions.sql
{{ config(
    materialized='incremental',
    unique_key=['transaction_id']
) }}

select
    t.transaction_id,
    t.portfolio_id,
    t.instrument_id,
    cast(format(t.transaction_date, 'yyyyMMdd') as int) as date_key,
    t.transaction_type,
    t.quantity,
    t.net_amount
from {{ ref('stg_transactions') }} t

{% if is_incremental() %}
where t.transaction_date >= dateadd(day, -5, (select max(convert(date, cast(date_key as varchar(8)), 112)) from {{ this }}))
{% endif %}

5.5 Tests#

# models/marts/schema.yml
version: 2

models:
  - name: dim_instrument
    columns:
      - name: instrument_id
        tests:
          - not_null
      - name: is_current
        tests:
          - accepted_values:
              values: ['true', 'false']

  - name: fact_positions
    columns:
      - name: portfolio_id
        tests:
          - not_null
          - relationships:
              to: ref('dim_portfolio')
              field: portfolio_id
      - name: instrument_id
        tests:
          - not_null
          - relationships:
              to: ref('dim_instrument')
              field: instrument_id
    tests:
      - unique:
          column_name: "portfolio_id || '-' || instrument_id || '-' || date_key"

  - name: fact_returns_daily
    columns:
      - name: portfolio_id
        tests:
          - not_null
          - relationships:
              to: ref('dim_portfolio')
              field: portfolio_id
      - name: benchmark_id
        tests:
          - relationships:
              to: ref('dim_benchmark')
              field: benchmark_id

  - name: fact_transactions
    columns:
      - name: transaction_id
        tests:
          - unique
          - not_null

5.6 profiles.yml#

# profiles.yml
investment_gold:
  target: prod
  outputs:
    prod:
      type: fabric
      driver: "ODBC Driver 18 for SQL Server"
      host: <wh_gold-sql-endpoint>.datawarehouse.fabric.microsoft.com
      port: 1433
      database: wh_gold
      schema: dbo
      authentication: ServicePrincipal
      tenant_id: "{{ env_var('FABRIC_TENANT_ID') }}"
      client_id: "{{ env_var('FABRIC_SP_CLIENT_ID') }}"
      client_secret: "{{ env_var('FABRIC_SP_CLIENT_SECRET') }}"
      threads: 4

Interactive development can swap authentication: ServicePrincipal for CLI-based Entra ID auth (az login); production/scheduled runs should use a service principal, which is Microsoft's stated recommendation for unattended dbt-fabric runs ^[S7].

5.7 Scheduling dbt build from an Airflow job in Data Factory#

Data Factory's Apache Airflow integration lets teams express orchestration as Python DAGs instead of the visual pipeline designer, which is the documented pattern for operationalizing scheduled dbt runs against Fabric — there is no native dbt scheduler inside Fabric itself ^{[S3] [S7]}.

# dags/gold_dbt_build.py
from airflow import DAG
from airflow.operators.bash import BashOperator
from datetime import datetime, timedelta

default_args = {"retries": 2, "retry_delay": timedelta(minutes=5)}

with DAG(
    dag_id="wh_gold_dbt_build",
    schedule_interval="0 */2 * * *",   # every two hours
    start_date=datetime(2026, 1, 1),
    catchup=False,
    default_args=default_args,
) as dag:

    dbt_build = BashOperator(
        task_id="dbt_build_wh_gold",
        bash_command="cd /opt/dbt/investment_gold && dbt build --profiles-dir . --target prod",
    )

This DAG runs inside Data Factory's Airflow integration, giving dbt build (which runs models and tests together) a recurring schedule and Airflow's retry/alerting semantics on top of the plain CLI invocation ^[S3].

6. Telemetry#

Two different telemetry needs exist in this pipeline, and they deliberately go to two different stores:

• sqldb_meta holds the coarse, run-level audit trail: one etl_run row per pipeline execution, with status, row counts, and timing — enough to answer "did last night's load succeed, and how many rows moved."
• eh_ops (an Eventhouse) holds granular, high-volume, timestamped operational events — one event per file processed, per API page fetched, per retry, per row-count checkpoint inside a Spark job — because Eventhouse is purpose-built for fast, indexed queries over exactly this kind of high-cardinality, time-ordered event stream at scale, independent of and complementary to the coarser SQL run log ^[S8].

There are two viable ways to get events into eh_ops from outside Fabric: routing through an Eventstream's custom app/endpoint destination, or writing directly to the KQL database with the Kusto ingestion client. This prototype shows the direct-Kusto path from an Azure Function, since it needs no intermediate Eventstream item for a prototype-scale telemetry volume; an Eventstream custom endpoint is the better choice if the same event stream should later fan out to multiple destinations (Eventhouse plus Activator plus a lakehouse) without changing the producer ^[S8].

# Azure Function (Python, HTTP-triggered) — direct Kusto ingestion into eh_ops.
import os
import json
import logging
import azure.functions as func
from azure.kusto.data import KustoConnectionStringBuilder
from azure.kusto.ingest import QueuedIngestClient, IngestionProperties

KUSTO_URI = os.environ["EH_OPS_QUERY_URI"]        # e.g. https://<eh_ops-cluster>.kusto.fabric.microsoft.com
DATABASE = "eh_ops"
TABLE = "PipelineEvents"

kcsb = KustoConnectionStringBuilder.with_aad_application_key_authentication(
    KUSTO_URI,
    os.environ["FABRIC_SP_CLIENT_ID"],
    os.environ["FABRIC_SP_CLIENT_SECRET"],
    os.environ["FABRIC_TENANT_ID"],
)
ingest_client = QueuedIngestClient(kcsb)


def main(req: func.HttpRequest) -> func.HttpResponse:
    body = req.get_json()
    event = {
        "EventTime": body["event_time"],
        "EntityName": body["entity_name"],
        "RunId": body["run_id"],
        "EventType": body["event_type"],      # e.g. 'FILE_PROCESSED', 'API_PAGE_FETCHED', 'RETRY'
        "RowCount": body.get("row_count"),
        "DurationMs": body.get("duration_ms"),
        "Detail": json.dumps(body.get("detail", {})),
    }

    ingestion_props = IngestionProperties(database=DATABASE, table=TABLE, data_format="JSON")
    ingest_client.ingest_from_stream(
        stream_descriptor=json.dumps(event),
        ingestion_properties=ingestion_props,
    )

    logging.info("Ingested %s event for run %s", event["EventType"], event["RunId"])
    return func.HttpResponse(status_code=202)

A representative KQL query over eh_ops for a health dashboard:

PipelineEvents
| where EventTime > ago(24h)
| summarize
    Events = count(),
    Failures = countif(EventType == "RETRY"),
    AvgDurationMs = avg(DurationMs)
  by EntityName, bin(EventTime, 1h)
| order by EventTime desc

7. Verify it worked#

A short checklist to confirm the prototype is producing data end-to-end, from source to gold:

1. Bronze landed — in a Fabric notebook against lh_analytics: spark.read.table("bronze_custodian_a_positions").count() returns a growing row count after each file-source pipeline run, and _source/_ingested_at are populated on every row.
2. Mirror shortcut is live — spark.read.table("lh_analytics.mirror_portfolio_shortcut.transactions").count() returns rows without ever running a pipeline for that source.
3. Silver conformed — select count(*), count(distinct portfolio_id || '-' || instrument_id || '-' || cast(position_date as varchar)) from silver_positions in the lakehouse SQL analytics endpoint returns equal counts (no duplicate business keys survived dedupe).
4. Gold built and tested — dbt build --target prod completes with all tests passing; check dbt_build.log or Airflow's task logs for the run.
5. T-SQL spot-check against wh_gold:

-- Row counts across the star schema — nothing should be zero after a full run.
select 'dim_instrument' as tbl, count(*) as row_count from dim_instrument
union all select 'fact_positions', count(*) from fact_positions
union all select 'fact_returns_daily', count(*) from fact_returns_daily
union all select 'fact_risk_measures', count(*) from fact_risk_measures
union all select 'fact_transactions', count(*) from fact_transactions;

-- Referential spot-check: every fact_positions row should join to a known instrument.
select top 10 fp.*
from fact_positions fp
left join dim_instrument di on fp.instrument_id = di.instrument_id
where di.instrument_id is null;   -- expect 0 rows

6. KQL spot-check against eh_ops:

PipelineEvents
| where EventTime > ago(1h)
| count

A non-zero count here, alongside a SUCCEEDED row in sqldb_meta.dbo.etl_run for the same window, confirms both telemetry paths and the control-plane audit trail are working together.

Internals#

Architecture & design#

Physically, this prototype is a chain of Delta tables: bronze and silver both live as managed Delta tables inside one lakehouse's OneLake storage, the mirrored source lands as its own Delta tables in OneLake and is joined in purely through a shortcut (a symbolic-link-like pointer that Fabric auto-registers as a table when it targets Delta Parquet data), and gold is a second, independent set of Delta tables physically owned by the warehouse engine but still published to OneLake in the same open format ^{[S1] [S6] [S2]}. This is why a single T-SQL statement can cross from a lakehouse SQL analytics endpoint into a warehouse or a mirrored database using three-part naming with zero data duplication — every layer, regardless of which Fabric item owns it, is Delta on OneLake underneath ^{[S5] [S2]}.

How it works internally#

Mirroring's replication path is a managed poller: for database and open mirroring, an internal replicator engine polls a Fabric landing zone at high frequency and merges incremental Delta files into the target table as they arrive, with backoff logic that relaxes polling frequency when source change volume is low and tightens it again when activity resumes — this is what allows change propagation as fast as roughly 15 seconds without a hand-authored CDC pipeline ^[S4]. On the warehouse side, Fabric's distributed SQL engine (Polaris, originally built for Azure Synapse and carried forward as the query-processing foundation for Fabric Warehouse) treats every table as a grid of data cells addressed by a user partition function and a system hash-distribution function; a two-phase Cascades-based optimizer picks a physical plan that minimizes data movement, injecting explicit "data move enforcer" operators (hash redistribution or broadcast) only where the plan's distribution requirements force it, and each resulting task is compiled back down to native T-SQL executed by a SQL Server instance on the compute node — combining big-data-style scale-out scheduling with SQL Server's mature scale-up columnar execution ^[S12]. On the Direct Lake side, Power BI never eagerly loads gold tables: columns are transcoded from OneLake into memory lazily on first query touch, and a metadata-only "framing" operation periodically rebinds the model to the Delta table's current version so subsequent queries see a consistent snapshot until the next frame ^[S11].

Performance characteristics#

Write-path performance in lh_analytics bronze and silver is governed by the same Delta file-layout mechanics as any Fabric lakehouse: Optimized Write shuffles data across executors before writing so each output partition is handled by one executor, cutting small-file counts dramatically on partitioned tables (one community benchmark measured a roughly 19x downstream query improvement after reducing 175,008 files to 1,823 on a heavily partitioned table), though the same source found Optimized Write can hurt a non-partitioned table because fewer, larger output files reduce read parallelism relative to the executor core count — a caution worth weighing before blanket-enabling it on every silver table in this design ^[S10]. V-Order is a complementary, separate write-time optimization: it's disabled by default in new workspaces, costs roughly 15% longer writes on average, and pays off specifically for read-heavy, repeated-scan patterns like gold tables serving BI — which argues for leaving it off on bronze/silver (write-heavy, transformation-heavy) and considering it, or a readHeavyforSpark resource profile, closer to the gold-facing tables in lh_analytics that Direct Lake or Power BI hit directly ^[S9]. On the warehouse side, the Polaris paper that underpins Fabric Warehouse's query engine reports scale characteristics from its own benchmarks — for example all 22 TPC-H queries completing at 1 PB across 420 execution nodes, and 5,000 concurrent TPC-DS sessions completing on just 10 nodes — which is evidence of the engine's designed scale envelope, not a claim about this prototype's own (much smaller) wh_gold workload ^[S12]. The dbt-fabric CTAS/DROP-CREATE materialization mechanic discussed in section 5.4 is itself a performance-relevant caveat: an "incremental" dbt run on this adapter is not guaranteed to be a cheap row-level merge, so fact_positions and fact_returns_daily build times should be measured directly as data volume grows, not assumed from experience with MERGE-native warehouse platforms ^[S7].