powerbi-fabric-pipelines

Skill: Microsoft Fabric — Data Pipelines & Ingestion

Trigger

Activate when the user mentions: data pipeline, ingestion, ETL, ELT, Fabric pipeline, Data Factory, copy activity, notebook, Spark, dataflow gen2, lakehouse, OneLake, data load, incremental load, watermark, CDC, change data capture, full load, delta load, pipeline orchestration

What You Know

You have designed and delivered enterprise data pipelines across Azure Data Factory, Synapse, and now Microsoft Fabric. You understand the full ingestion stack from source connectors to Delta table landing, and you know every trick for incremental loading, CDC, and pipeline reliability.

Ingestion Strategy Decision Tree

Source data changes how?
├── Full refresh (small tables <1M rows)   → Copy Activity, full load
├── Append-only (logs, events, IoT)        → Streaming / Eventstream
├── Inserts + Updates (CDC available)      → CDC with watermark
├── Inserts + Updates (no CDC)             → Watermark on modified_date
└── Deletes present                        → Full load OR CDC (SCD Type 2 in Silver)

Bronze Layer Ingestion Patterns

Pattern 1: Full Load (Small Dimensions)

# Fabric Notebook — full load to Bronze Delta
from pyspark.sql import SparkSession

spark = SparkSession.builder.getOrCreate()

df = spark.read.format("jdbc") \
    .option("url", "jdbc:sqlserver://server:1433;database=CRM") \
    .option("dbtable", "dbo.Customer") \
    .option("user", "{secret}") \
    .option("password", "{secret}") \
    .load()

# Add metadata
from pyspark.sql.functions import current_timestamp, lit
df = df \
    .withColumn("_ingested_at", current_timestamp()) \
    .withColumn("_source_system", lit("CRM"))

# Write to Bronze (overwrite for full load)
df.write.format("delta") \
    .mode("overwrite") \
    .option("overwriteSchema", "true") \
    .save("abfss://[email protected]/workspace/Tables/raw_crm_customer")

Pattern 2: Watermark Incremental Load

# Read last watermark from control table
watermark_df = spark.read.format("delta").load("abfss://bronze@.../control/watermarks")
last_watermark = watermark_df.filter("table_name = 'orders'").collect()[0]["last_value"]

# Load only new/changed records
df = spark.read.format("jdbc") \
    .option("dbtable", f"(SELECT * FROM dbo.Orders WHERE modified_date > '{last_watermark}') t") \
    .load()

# Merge into Bronze Delta (UPSERT)
from delta.tables import DeltaTable

target = DeltaTable.forPath(spark, "abfss://bronze@.../Tables/raw_erp_orders")
target.alias("t").merge(
    df.alias("s"),
    "t.order_id = s.order_id"
).whenMatchedUpdateAll() \
 .whenNotMatchedInsertAll() \
 .execute()

# Update watermark
new_watermark = df.agg({"modified_date": "max"}).collect()[0][0]
# ... update control table