{ "nbformat": 4, "nbformat_minor": 5, "metadata": { "kernelspec": { "display_name": "Python 3", "language": "python", "name": "python3" }, "language_info": { "name": "python", "version": "3.13.0" }, "blog_metadata": { "topic": "What Fabric Dataflows Gen2 Means for the Next Phase of Microsoft Analytics Engineering", "slug": "what-fabric-dataflows-gen2-means-for-the-next-phase-of-micro", "generated_by": "LinkedIn Post Generator + Azure OpenAI", "generated_at": "2026-08-05T19:15:07.016Z" } }, "cells": [ { "cell_type": "markdown", "metadata": {}, "source": [ "# What Fabric Dataflows Gen2 Means for the Next Phase of Microsoft Analytics Engineering\n", "\n", "This notebook turns the blog post into a hands-on validation workflow. The focus is not just on Dataflows Gen2 as a UI feature, but on its role as a governed transformation layer inside Microsoft Fabric. We will simulate ingestion, transformation, quality checks, environment promotion, and incremental processing patterns using Python so the operating-model ideas can be tested directly." ] }, { "cell_type": "code", "metadata": {}, "source": [ "%pip install -q pandas pyarrow matplotlib" ], "execution_count": null, "outputs": [] }, { "cell_type": "code", "metadata": {}, "source": [ "import json\n", "from pathlib import Path\n", "from textwrap import dedent\n", "\n", "import pandas as pd\n", "import matplotlib.pyplot as plt" ], "execution_count": null, "outputs": [] }, { "cell_type": "markdown", "metadata": {}, "source": [ "## Platform signal vs feature story\n", "\n", "The blog argues that Dataflows Gen2 matters because it standardizes transformation in a governed, reusable way. This cell renders a simple architecture view of the intended operating model using Python so the flow can be inspected in notebook form." ] }, { "cell_type": "code", "metadata": {}, "source": [ "architecture = {\n", " \"Operational Sources\": [\"SQL\", \"CSV\", \"APIs\"],\n", " \"Fabric Dataflows Gen2\": [\"Power Query at scale\"],\n", " \"Lakehouse / OneLake Bronze\": [],\n", " \"Warehouse / Lakehouse Silver\": [],\n", " \"Semantic Model Gold\": [],\n", " \"Power BI Reports\": [],\n", " \"Data Activator / Monitoring\": []\n", "}\n", "\n", "edges = [\n", " (\"Operational Sources\", \"Fabric Dataflows Gen2\"),\n", " (\"Fabric Dataflows Gen2\", \"Lakehouse / OneLake Bronze\"),\n", " (\"Lakehouse / OneLake Bronze\", \"Warehouse / Lakehouse Silver\"),\n", " (\"Warehouse / Lakehouse Silver\", \"Semantic Model Gold\"),\n", " (\"Semantic Model Gold\", \"Power BI Reports\"),\n", " (\"Fabric Dataflows Gen2\", \"Data Activator / Monitoring\")\n", "]\n", "\n", "print(\"Architecture nodes:\")\n", "for node, details in architecture.items():\n", " print(f\"- {node}: {', '.join(details) if details else 'n/a'}\")\n", "\n", "print(\"\\nFlow:\")\n", "for src, dst in edges:\n", " print(f\"{src} -> {dst}\")" ], "execution_count": null, "outputs": [] }, { "cell_type": "markdown", "metadata": {}, "source": [ "## Ingestion pattern: normalize and land predictable output\n", "\n", "This example mirrors a Dataflows Gen2-style source-to-destination step. It creates a sample CSV, normalizes column names, stamps ingestion time, and writes a parquet output to simulate a Bronze landing pattern." ] }, { "cell_type": "code", "metadata": {}, "source": [ "import pandas as pd\n", "from pathlib import Path\n", "\n", "\n", "def ingest_csv_to_parquet(source_path: str, destination_path: str) -> None:\n", " df = pd.read_csv(source_path)\n", " df.columns = [c.strip().lower().replace(\" \", \"_\") for c in df.columns]\n", " df[\"ingested_utc\"] = pd.Timestamp.utcnow()\n", " df.to_parquet(destination_path, index=False)\n", "\n", "\n", "sample_orders = pd.DataFrame([\n", " {\"Order ID\": 1, \"Customer ID\": 101, \"Order Date\": \"2026-01-15\", \"Amount\": \"120.50\", \"Status\": \"shipped\"},\n", " {\"Order ID\": 2, \"Customer ID\": 102, \"Order Date\": \"2026-01-16\", \"Amount\": \"89.99\", \"Status\": \"processing\"}\n", "])\n", "\n", "source_path = Path(\"sales_orders.csv\")\n", "destination_path = Path(\"bronze_sales_orders.parquet\")\n", "sample_orders.to_csv(source_path, index=False)\n", "\n", "ingest_csv_to_parquet(str(source_path), str(destination_path))\n", "\n", "bronze_df = pd.read_parquet(destination_path)\n", "print(bronze_df)\n", "print(\"\\nColumns:\", list(bronze_df.columns))" ], "execution_count": null, "outputs": [] }, { "cell_type": "markdown", "metadata": {}, "source": [ "## Reusable transformation logic\n", "\n", "The next pattern represents the kind of repeatable cleanup logic that should be reviewable and reusable across teams. It applies type coercion, null handling, conformance, and a derived business field." ] }, { "cell_type": "code", "metadata": {}, "source": [ "import pandas as pd\n", "\n", "\n", "def transform_orders(df: pd.DataFrame) -> pd.DataFrame:\n", " clean = df.copy()\n", " clean[\"order_date\"] = pd.to_datetime(clean[\"order_date\"], errors=\"coerce\")\n", " clean[\"amount\"] = pd.to_numeric(clean[\"amount\"], errors=\"coerce\").fillna(0)\n", " clean[\"status\"] = clean[\"status\"].fillna(\"Unknown\").astype(str).str.title()\n", " clean = clean.dropna(subset=[\"customer_id\", \"order_date\"])\n", " clean[\"order_year_month\"] = clean[\"order_date\"].dt.strftime(\"%Y-%m\")\n", " return clean\n", "\n", "\n", "sample = pd.DataFrame([\n", " {\"customer_id\": 101, \"order_date\": \"2026-01-15\", \"amount\": \"120.50\", \"status\": \"shipped\"},\n", " {\"customer_id\": 102, \"order_date\": \"bad-date\", \"amount\": \"15\", \"status\": None},\n", " {\"customer_id\": None, \"order_date\": \"2026-01-20\", \"amount\": \"oops\", \"status\": \"returned\"}\n", "])\n", "\n", "transformed = transform_orders(sample)\n", "print(transformed)" ], "execution_count": null, "outputs": [] }, { "cell_type": "markdown", "metadata": {}, "source": [ "## Sequence of governed transformation\n", "\n", "This cell converts the sequence-diagram idea into a simple executable event log. It helps validate the intended order of operations from source extraction through BI consumption." ] }, { "cell_type": "code", "metadata": {}, "source": [ "sequence_steps = [\n", " (\"Source System\", \"Dataflows Gen2\", \"Extract data\"),\n", " (\"Dataflows Gen2\", \"Dataflows Gen2\", \"Apply Power Query transformations\"),\n", " (\"Dataflows Gen2\", \"OneLake\", \"Write curated tables/files\"),\n", " (\"OneLake\", \"Warehouse\", \"Load modeled analytics layer\"),\n", " (\"Warehouse\", \"BI Model\", \"Serve semantic model\"),\n", " (\"BI Model\", \"Source System\", \"Business insights and feedback loop\")\n", "]\n", "\n", "sequence_df = pd.DataFrame(sequence_steps, columns=[\"from\", \"to\", \"action\"])\n", "print(sequence_df.to_string(index=False))" ], "execution_count": null, "outputs": [] }, { "cell_type": "markdown", "metadata": {}, "source": [ "## Refresh workflow pattern in Python\n", "\n", "The original post included a PowerShell refresh example. Because this notebook uses Python, the same idea is represented here as a refresh request payload that could be logged, queued, or sent to an API in a real implementation." ] }, { "cell_type": "code", "metadata": {}, "source": [ "import pandas as pd\n", "import json\n", "\n", "\n", "def build_refresh_request(workspace_name: str = \"Analytics-Engineering\", dataflow_name: str = \"Sales-Orders-Gen2\") -> dict:\n", " return {\n", " \"workspace\": workspace_name,\n", " \"dataflow\": dataflow_name,\n", " \"triggerAt\": pd.Timestamp.utcnow().isoformat(),\n", " \"reason\": \"Scheduled refresh for curated ingestion\"\n", " }\n", "\n", "\n", "refresh_request = build_refresh_request()\n", "print(json.dumps(refresh_request, indent=2))" ], "execution_count": null, "outputs": [] }, { "cell_type": "markdown", "metadata": {}, "source": [ "## Data quality checks after landing\n", "\n", "A key argument in the blog is that refresh success is not the same as trustworthy data. This example runs simple quality checks immediately after landing so failures can be tied to owners and thresholds." ] }, { "cell_type": "code", "metadata": {}, "source": [ "import pandas as pd\n", "\n", "\n", "def validate_orders(df: pd.DataFrame) -> dict:\n", " return {\n", " \"row_count\": int(len(df)),\n", " \"null_customer_id\": int(df[\"customer_id\"].isna().sum()),\n", " \"negative_amounts\": int((df[\"amount\"] < 0).sum()),\n", " \"duplicate_order_ids\": int(df[\"order_id\"].duplicated().sum()),\n", " }\n", "\n", "\n", "orders = pd.DataFrame([\n", " {\"order_id\": 1, \"customer_id\": 101, \"amount\": 25.0},\n", " {\"order_id\": 2, \"customer_id\": None, \"amount\": -5.0},\n", " {\"order_id\": 2, \"customer_id\": 103, \"amount\": 10.0},\n", "])\n", "\n", "quality_report = validate_orders(orders)\n", "print(quality_report)" ], "execution_count": null, "outputs": [] }, { "cell_type": "markdown", "metadata": {}, "source": [ "## Environment-driven promotion across dev, test, and prod\n", "\n", "The blog stresses that common tooling does not replace operating discipline. This Python version of the environment configuration pattern shows how a governed promotion path can select the right workspace and lakehouse target." ] }, { "cell_type": "code", "metadata": {}, "source": [ "def select_environment_config(environment: str = \"prod\") -> dict:\n", " config = {\n", " \"dev\": {\"Lakehouse\": \"lh_dev\", \"Workspace\": \"ws_dev\"},\n", " \"test\": {\"Lakehouse\": \"lh_test\", \"Workspace\": \"ws_test\"},\n", " \"prod\": {\"Lakehouse\": \"lh_prod\", \"Workspace\": \"ws_prod\"},\n", " }\n", " selected = config[environment]\n", " return {\n", " \"Environment\": environment,\n", " \"Workspace\": selected[\"Workspace\"],\n", " \"Lakehouse\": selected[\"Lakehouse\"],\n", " }\n", "\n", "\n", "env_config = select_environment_config(\"prod\")\n", "print(json.dumps(env_config, indent=2))" ], "execution_count": null, "outputs": [] }, { "cell_type": "markdown", "metadata": {}, "source": [ "## Operating model flow for analytics engineering\n", "\n", "This cell turns the final flowchart into a structured list of stages. It reinforces the idea that Dataflows Gen2 should sit inside a broader governed lifecycle, not as an isolated feature." ] }, { "cell_type": "code", "metadata": {}, "source": [ "operating_flow = [\n", " \"Analytics Engineer\",\n", " \"Build transformation in Dataflows Gen2\",\n", " \"Land standardized data in OneLake\",\n", " \"Apply quality checks and contracts\",\n", " \"Publish to Warehouse or Semantic Model\",\n", " \"Self-service BI with governed reuse\"\n", "]\n", "\n", "for i, step in enumerate(operating_flow, start=1):\n", " print(f\"{i}. {step}\")" ], "execution_count": null, "outputs": [] }, { "cell_type": "markdown", "metadata": {}, "source": [ "## Incremental processing pattern\n", "\n", "The blog closes with the idea that the next phase of analytics engineering requires repeatable, governed patterns. Incremental filtering is one such pattern, and this example validates a simple watermark-based approach." ] }, { "cell_type": "code", "metadata": {}, "source": [ "import pandas as pd\n", "\n", "\n", "def filter_incremental(df: pd.DataFrame, watermark: str) -> pd.DataFrame:\n", " ts = pd.to_datetime(df[\"last_modified_utc\"], errors=\"coerce\", utc=True)\n", " watermark_ts = pd.Timestamp(watermark)\n", " if watermark_ts.tzinfo is None:\n", " watermark_ts = watermark_ts.tz_localize(\"UTC\")\n", " return df.loc[ts > watermark_ts].copy()\n", "\n", "\n", "changes = pd.DataFrame([\n", " {\"order_id\": 1, \"last_modified_utc\": \"2026-02-01T10:00:00Z\"},\n", " {\"order_id\": 2, \"last_modified_utc\": \"2026-02-03T08:30:00Z\"},\n", " {\"order_id\": 3, \"last_modified_utc\": \"2026-02-04T12:15:00Z\"},\n", "])\n", "\n", "incremental = filter_incremental(changes, \"2026-02-02T00:00:00Z\")\n", "print(incremental)" ], "execution_count": null, "outputs": [] }, { "cell_type": "markdown", "metadata": {}, "source": [ "## Contract-first validation\n", "\n", "To make the blog's governance argument concrete, this cell defines a minimal transformation contract and checks whether a sample asset satisfies it. This helps shift the conversation from tool preference to operating discipline." ] }, { "cell_type": "code", "metadata": {}, "source": [ "required_contract_fields = [\n", " \"named_inputs_outputs\",\n", " \"declared_owner\",\n", " \"refresh_expectation\",\n", " \"data_quality_expectations\",\n", " \"consumer_alignment\",\n", "]\n", "\n", "sample_contract = {\n", " \"named_inputs_outputs\": {\n", " \"sources\": [\"sales_orders.csv\", \"sap_orders_feed\"],\n", " \"destination\": \"lh_prod.bronze_sales_orders\",\n", " \"business_purpose\": \"Curated order ingestion for downstream semantic models\"\n", " },\n", " \"declared_owner\": \"Sales Analytics Engineering\",\n", " \"refresh_expectation\": \"daily\",\n", " \"data_quality_expectations\": {\n", " \"null_customer_id_max\": 0,\n", " \"negative_amounts_max\": 0,\n", " \"duplicate_order_ids_max\": 0\n", " },\n", " \"consumer_alignment\": [\"sales_semantic_model\", \"margin_report\"]\n", "}\n", "\n", "validation = {field: field in sample_contract and sample_contract[field] not in [None, \"\", [], {}] for field in required_contract_fields}\n", "print(\"Contract validation:\")\n", "print(json.dumps(validation, indent=2))\n", "print(\"\\nAll required fields present:\", all(validation.values()))" ], "execution_count": null, "outputs": [] }, { "cell_type": "markdown", "metadata": {}, "source": [ "## Capacity planning illustration\n", "\n", "The post highlights Dataflow Gen2 parallel task limits as an executive planning input. This simple calculation compares scheduled workloads against capacity bands to show how overlapping refreshes can become an operating issue." ] }, { "cell_type": "code", "metadata": {}, "source": [ "capacity_limits = {\n", " \"F2-F32\": 96,\n", " \"F64-F512\": 384\n", "}\n", "\n", "scheduled_dataflows = pd.DataFrame([\n", " {\"domain\": \"sales\", \"parallel_tasks_needed\": 20},\n", " {\"domain\": \"finance\", \"parallel_tasks_needed\": 18},\n", " {\"domain\": \"supply_chain\", \"parallel_tasks_needed\": 25},\n", " {\"domain\": \"manufacturing\", \"parallel_tasks_needed\": 22},\n", " {\"domain\": \"customer\", \"parallel_tasks_needed\": 19},\n", "])\n", "\n", "total_parallel_tasks = int(scheduled_dataflows[\"parallel_tasks_needed\"].sum())\n", "print(\"Total scheduled parallel tasks:\", total_parallel_tasks)\n", "for band, limit in capacity_limits.items():\n", " print(f\"{band}: limit={limit}, within_limit={total_parallel_tasks <= limit}\")\n", "\n", "scheduled_dataflows.plot(kind=\"bar\", x=\"domain\", y=\"parallel_tasks_needed\", legend=False, title=\"Parallel Tasks Needed by Domain\")\n", "plt.ylabel(\"Parallel tasks\")\n", "plt.tight_layout()\n", "plt.show()" ], "execution_count": null, "outputs": [] }, { "cell_type": "markdown", "metadata": {}, "source": [ "## Summary\n", "\n", "This notebook validated the blog's main claim: Dataflows Gen2 is most important as a governed transformation surface, not just as a faster self-service ETL feature. The practical patterns above showed how ingestion, reusable transformations, quality checks, environment promotion, contracts, and capacity planning all contribute to a shared operating layer.\n", "\n", "## Next Steps\n", "\n", "1. Define a transformation contract for every domain.\n", "2. Make Dataflows Gen2 the default for understandable, repeatable ingestion and cleanup.\n", "3. Keep notebook and specialized pipeline paths available for workloads that truly need code-first engineering.\n", "4. Track platform outcomes such as duplicate logic reduction, ownership clarity, semantic-model stability, and predictable capacity behavior." ] } ] }