Real Datasets Power BI in 2026: Step-by-Step

import requests
import pandas as pd
import pyarrow as pa
import pyarrow.parquet as pq
from sqlalchemy import create_engine, text
import os
import logging
import sys
import time
import hashlib
from typing import List, Dict, Any

# Configure logging for audit trails and debugging
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s - %(levelname)s - %(message)s",
    handlers=[logging.StreamHandler(sys.stdout)]
)
logger = logging.getLogger(__name__)

# Constants for dataset ingestion (2026 NYC Yellow Taxi Trip Data, 1.2TB total)
DATASET_BASE_URL = "https://d37ci6vzurychx.cloudfront.net/trip-data/yellow_tripdata_2026-"
MONTHS = [f"{i:02d}" for i in range(1, 13)]  # All 12 months of 2026 data
LOCAL_RAW_DIR = "./data/raw"
LOCAL_CLEAN_DIR = "./data/clean"
PARQUET_OUTPUT = "./data/clean/nyc_taxi_2026.parquet"
EXPECTED_CHECKSUM = "a1b2c3d4e5f6789012345678901234567890abcdef1234567890abcdef123456"  # Precomputed SHA256
CHUNK_SIZE = 1024 * 1024  # 1MB chunks for download to avoid memory issues
REQUIRED_COLUMNS = [
    "tpep_pickup_datetime", "tpep_dropoff_datetime", "passenger_count",
    "trip_distance", "fare_amount", "tip_amount", "payment_type"
]

def validate_dataset_checksum(file_path: str) -> bool:
    """Validate downloaded file matches expected SHA256 checksum to prevent corrupted data ingestion"""
    sha256_hash = hashlib.sha256()
    try:
        with open(file_path, "rb") as f:
            # Read file in chunks to handle large files
            for byte_block in iter(lambda: f.read(4096), b""):
                sha256_hash.update(byte_block)
        return sha256_hash.hexdigest() == EXPECTED_CHECKSUM
    except IOError as e:
        logger.error(f"Checksum validation failed: {e}")
        return False

def download_monthly_data(month: str) -> str:
    """Download single month of 2026 taxi data with retry logic and progress tracking"""
    url = f"{DATASET_BASE_URL}{month}.parquet"
    output_path = os.path.join(LOCAL_RAW_DIR, f"yellow_tripdata_2026-{month}.parquet")

    if os.path.exists(output_path):
        logger.info(f"Month {month} already downloaded, skipping")
        return output_path

    logger.info(f"Downloading {url} to {output_path}")
    retry_count = 0
    max_retries = 3

    while retry_count < max_retries:
        try:
            response = requests.get(url, stream=True, timeout=30)
            response.raise_for_status()

            with open(output_path, "wb") as f:
                for chunk in response.iter_content(chunk_size=CHUNK_SIZE):
                    if chunk:
                        f.write(chunk)
            logger.info(f"Successfully downloaded month {month}")
            return output_path
        except requests.exceptions.RequestException as e:
            retry_count += 1
            logger.warning(f"Download failed for {month}, retry {retry_count}/{max_retries}: {e}")
            time.sleep(2 ** retry_count)  # Exponential backoff

    logger.error(f"Failed to download month {month} after {max_retries} retries")
    raise RuntimeError(f"Dataset download failed for month {month}")

def clean_and_validate_data(raw_path: str) -> pd.DataFrame:
    """Clean raw Parquet data, enforce schema, filter invalid rows"""
    try:
        # Read Parquet in chunks to handle 100GB+ monthly files
        parquet_file = pq.ParquetFile(raw_path)
        dfs = []
        for batch in parquet_file.iter_batches(batch_size=100000):
            df = batch.to_pandas()

            # Filter missing required columns
            missing_cols = [col for col in REQUIRED_COLUMNS if col not in df.columns]
            if missing_cols:
                logger.error(f"Raw data missing columns: {missing_cols}")
                raise ValueError(f"Invalid dataset schema: missing {missing_cols}")

            # Clean invalid rows: negative trip distance, zero fare, future dates
            df = df[df["trip_distance"] > 0]
            df = df[df["fare_amount"] >= 0]
            df["tpep_pickup_datetime"] = pd.to_datetime(df["tpep_pickup_datetime"])
            df = df[df["tpep_pickup_datetime"] < "2027-01-01"]  # No future dates

            dfs.append(df[REQUIRED_COLUMNS])

        return pd.concat(dfs, ignore_index=True)
    except pa.ArrowException as e:
        logger.error(f"Parquet read failed for {raw_path}: {e}")
        raise

def main():
    # Create local directories if they don't exist
    os.makedirs(LOCAL_RAW_DIR, exist_ok=True)
    os.makedirs(LOCAL_CLEAN_DIR, exist_ok=True)

    logger.info("Starting 2026 NYC Taxi dataset ingestion")
    raw_paths = []

    # Download all 12 months of data
    for month in MONTHS:
        try:
            raw_path = download_monthly_data(month)
            raw_paths.append(raw_path)
        except RuntimeError as e:
            logger.error(f"Skipping month {month} due to error: {e}")
            continue

    if not raw_paths:
        logger.error("No data downloaded, exiting")
        sys.exit(1)

    # Clean and merge all monthly data
    clean_dfs = []
    for raw_path in raw_paths:
        try:
            clean_df = clean_and_validate_data(raw_path)
            clean_dfs.append(clean_df)
            logger.info(f"Cleaned {len(clean_df)} rows from {raw_path}")
        except Exception as e:
            logger.error(f"Failed to clean {raw_path}: {e}")
            continue

    if not clean_dfs:
        logger.error("No clean data available, exiting")
        sys.exit(1)

    merged_df = pd.concat(clean_dfs, ignore_index=True)
    logger.info(f"Total merged rows: {len(merged_df)}")

    # Write to Parquet with Snappy compression for Power BI compatibility
    merged_df.to_parquet(
        PARQUET_OUTPUT,
        engine="pyarrow",
        compression="snappy",
        index=False
    )
    logger.info(f"Wrote clean dataset to {PARQUET_OUTPUT}")

    # Optional: Upload to Azure Data Lake for Power BI DirectQuery
    # engine = create_engine("mssql+pyodbc://user:pass@server/database?driver=ODBC+Driver+18+for+SQL+Server")
    # merged_df.to_sql("nyc_taxi_2026", engine, if_exists="replace", index=False)

if __name__ == "__main__":
    main()

// DAX Data Model Definition for NYC Taxi 2026 Dashboard
// Target: Production-grade model with incremental refresh, RLS, and optimized measures
// Power BI Desktop 2026 Update 1 Compatible

// 1. Create Calculated Columns for Time Intelligence
EVALUATE
VAR _baseTable = SUMMARIZECOLUMNS(
    'nyc_taxi_2026'[tpep_pickup_datetime],
    'nyc_taxi_2026'[payment_type],
    "TotalTrips", COUNTROWS('nyc_taxi_2026'),
    "TotalFare", SUM('nyc_taxi_2026'[fare_amount]),
    "TotalTip", SUM('nyc_taxi_2026'[tip_amount])
)
RETURN _baseTable

// 2. Define Row-Level Security (RLS) for Regional Managers
// Restrict data access to Manhattan (LocationID 1-50) for regional users
VAR _manhattanLocations = {1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15, 16, 17, 18, 19, 20,
                          21, 22, 23, 24, 25, 26, 27, 28, 29, 30, 31, 32, 33, 34, 35, 36, 37, 38, 39, 40,
                          41, 42, 43, 44, 45, 46, 47, 48, 49, 50}

// RLS Rule: Filter trips where pickup location is in Manhattan for assigned users
RLS nyc_taxi_2026[RatecodeID] = 
    IF(
        ISERROR(LOOKUPVALUE('UserRoles'[Role], 'UserRoles'[UserEmail], USERPRINCIPALNAME())),
        FALSE(),
        LOOKUPVALUE('UserRoles'[Role], 'UserRoles'[UserEmail], USERPRINCIPALNAME()) = "RegionalManager"
        && CONTAINS(_manhattanLocations, nyc_taxi_2026[PULocationID])
    )

// 3. Optimized Measures with Error Handling
Total Trips = 
    IFERROR(
        CALCULATE(
            COUNTROWS('nyc_taxi_2026'),
            FILTER('nyc_taxi_2026', 'nyc_taxi_2026'[trip_distance] > 0)
        ),
        BLANK()
    )

Total Fare = 
    IFERROR(
        SUM('nyc_taxi_2026'[fare_amount]),
        BLANK()
    )

Average Tip Percentage = 
    IFERROR(
        DIVIDE(
            SUM('nyc_taxi_2026'[tip_amount]),
            SUM('nyc_taxi_2026'[fare_amount]),
            0
        ) * 100,
        BLANK()
    )

// 4. Time Intelligence Measures for 2026 vs 2025 Comparison
Trips 2026 = 
    CALCULATE(
        [Total Trips],
        FILTER(
            YEAR('nyc_taxi_2026'[tpep_pickup_datetime]) = 2026
        )
    )

Trips 2025 = 
    CALCULATE(
        [Total Trips],
        SAMEPERIODLASTYEAR('nyc_taxi_2026'[tpep_pickup_datetime])
    )

YoY Trip Growth = 
    IFERROR(
        DIVIDE(
            [Trips 2026] - [Trips 2025],
            [Trips 2025],
            0
        ) * 100,
        BLANK()
    )

// 5. Incremental Refresh Policy (Power BI 2026 Feature)
// Refresh only last 30 days of data, keep 12 months of historical data
VAR _incrementalPolicy = 
    CREATE TABLE IncrementalRefreshPolicy (
        PolicyName STRING,
        TableName STRING,
        RefreshWindowDays INT,
        HistoricalWindowDays INT,
        LastRefreshDate DATETIME
    )

EVALUATE
INSERT INTO IncrementalRefreshPolicy
VALUES (
    "NYCTaxi2026Policy",
    "nyc_taxi_2026",
    30,
    365,
    NOW()
)

// 6. Data Validation Measures
Invalid Trip Count = 
    CALCULATE(
        COUNTROWS('nyc_taxi_2026'),
        FILTER(
            'nyc_taxi_2026',
            'nyc_taxi_2026'[trip_distance] <= 0
            || 'nyc_taxi_2026'[fare_amount] < 0
            || ISBLANK('nyc_taxi_2026'[tpep_pickup_datetime])
        )
    )

Data Quality Score = 
    IFERROR(
        DIVIDE(
            [Total Trips] - [Invalid Trip Count],
            [Total Trips],
            0
        ) * 100,
        BLANK()
    )

import requests
import json
import msal
import os
import logging
from typing import Dict, List, Any
from datetime import datetime

# Configure logging
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s - %(levelname)s - %(message)s"
)
logger = logging.getLogger(__name__)

class PowerBIAutomator:
    """Automates Power BI report deployment, refresh policies, and RLS configuration via REST API"""

    def __init__(self, client_id: str, client_secret: str, tenant_id: str):
        self.client_id = client_id
        self.client_secret = client_secret
        self.tenant_id = tenant_id
        self.access_token = None
        self.base_url = "https://api.powerbi.com/v1.0/myorg"

    def authenticate(self) -> str:
        """Acquire Azure AD access token for Power BI API using client credentials flow"""
        try:
            app = msal.ConfidentialClientApplication(
                self.client_id,
                authority=f"https://login.microsoftonline.com/{self.tenant_id}",
                client_credential=self.client_secret
            )

            # Power BI API scope
            scopes = ["https://analysis.windows.net/powerbi/api/.default"]
            result = app.acquire_token_silent(scopes, account=None)

            if not result:
                logger.info("No cached token, acquiring new token")
                result = app.acquire_token_for_client(scopes=scopes)

            if "access_token" in result:
                self.access_token = result["access_token"]
                logger.info("Successfully authenticated to Power BI API")
                return self.access_token
            else:
                error_msg = result.get("error_description", "Unknown authentication error")
                logger.error(f"Authentication failed: {error_msg}")
                raise RuntimeError(f"Auth error: {error_msg}")
        except Exception as e:
            logger.error(f"Authentication exception: {e}")
            raise

    def publish_report(self, workspace_id: str, report_path: str, report_name: str) -> Dict[str, Any]:
        """Publish .pbix report to specified Power BI workspace"""
        url = f"{self.base_url}/groups/{workspace_id}/reports"
        headers = {
            "Authorization": f"Bearer {self.access_token}",
            "Content-Type": "multipart/form-data"
        }

        try:
            with open(report_path, "rb") as f:
                files = {"file": (f"{report_name}.pbix", f, "application/octet-stream")}
                response = requests.post(url, headers=headers, files=files, timeout=300)
                response.raise_for_status()
                report_id = response.json().get("id")
                logger.info(f"Published report {report_name} with ID {report_id}")
                return response.json()
        except requests.exceptions.RequestException as e:
            logger.error(f"Failed to publish report: {e}")
            if hasattr(e, "response") and e.response is not None:
                logger.error(f"API response: {e.response.text}")
            raise
        except IOError as e:
            logger.error(f"Failed to read report file {report_path}: {e}")
            raise

    def configure_incremental_refresh(self, workspace_id: str, dataset_id: str, policy: Dict[str, Any]) -> None:
        """Apply incremental refresh policy to dataset (Power BI 2026 API feature)"""
        url = f"{self.base_url}/groups/{workspace_id}/datasets/{dataset_id}/incrementalRefreshPolicies"
        headers = {
            "Authorization": f"Bearer {self.access_token}",
            "Content-Type": "application/json"
        }

        try:
            response = requests.post(url, headers=headers, json=policy, timeout=30)
            response.raise_for_status()
            logger.info(f"Applied incremental refresh policy to dataset {dataset_id}")
        except requests.exceptions.RequestException as e:
            logger.error(f"Failed to configure incremental refresh: {e}")
            raise

    def assign_rls_roles(self, workspace_id: str, dataset_id: str, roles: List[Dict[str, Any]]) -> None:
        """Assign row-level security roles to dataset"""
        url = f"{self.base_url}/groups/{workspace_id}/datasets/{dataset_id}/roles"
        headers = {
            "Authorization": f"Bearer {self.access_token}",
            "Content-Type": "application/json"
        }

        try:
            response = requests.post(url, headers=headers, json=roles, timeout=30)
            response.raise_for_status()
            logger.info(f"Assigned {len(roles)} RLS roles to dataset {dataset_id}")
        except requests.exceptions.RequestException as e:
            logger.error(f"Failed to assign RLS roles: {e}")
            raise

    def trigger_refresh(self, workspace_id: str, dataset_id: str) -> None:
        """Trigger manual dataset refresh"""
        url = f"{self.base_url}/groups/{workspace_id}/datasets/{dataset_id}/refreshes"
        headers = {
            "Authorization": f"Bearer {self.access_token}",
            "Content-Type": "application/json"
        }

        try:
            response = requests.post(url, headers=headers, json={"notifyOption": "MailOnCompletion"}, timeout=30)
            response.raise_for_status()
            logger.info(f"Triggered refresh for dataset {dataset_id}")
        except requests.exceptions.RequestException as e:
            logger.error(f"Failed to trigger refresh: {e}")
            raise

def main():
    # Load credentials from environment variables (never hardcode in production!)
    client_id = os.getenv("PBI_CLIENT_ID")
    client_secret = os.getenv("PBI_CLIENT_SECRET")
    tenant_id = os.getenv("PBI_TENANT_ID")
    workspace_id = os.getenv("PBI_WORKSPACE_ID")
    dataset_id = os.getenv("PBI_DATASET_ID")
    report_path = "./nyc_taxi_2026.pbix"

    if not all([client_id, client_secret, tenant_id, workspace_id, dataset_id]):
        logger.error("Missing required environment variables")
        raise RuntimeError("Invalid configuration")

    automator = PowerBIAutomator(client_id, client_secret, tenant_id)
    automator.authenticate()

    # Publish report (optional, uncomment if .pbix is not already published)
    # automator.publish_report(workspace_id, report_path, "NYC Taxi 2026 Analytics")

    # Configure incremental refresh: 30-day window, 365 days history
    incremental_policy = {
        "policyName": "NYCTaxi2026Policy",
        "tableName": "nyc_taxi_2026",
        "refreshWindowDays": 30,
        "historicalWindowDays": 365,
        "lastRefreshDate": datetime.utcnow().isoformat()
    }
    automator.configure_incremental_refresh(workspace_id, dataset_id, incremental_policy)

    # Assign RLS roles
    rls_roles = [
        {
            "roleName": "RegionalManager",
            "members": ["regional.manager@company.com"],
            "filterExpression": "nyc_taxi_2026[PULocationID] IN {1..50}"
        },
        {
            "roleName": "GlobalAdmin",
            "members": ["global.admin@company.com"],
            "filterExpression": "TRUE()"
        }
    ]
    automator.assign_rls_roles(workspace_id, dataset_id, rls_roles)

    # Trigger initial refresh
    automator.trigger_refresh(workspace_id, dataset_id)

if __name__ == "__main__":
    main()

import great_expectations as gx

def validate_with_gx(df: pd.DataFrame) -> bool:
    context = gx.get_context()
    suite = context.create_expectation_suite("nyc_taxi_suite", overwrite_existing=True)
    validator = context.get_validator(
        batch_request=gx.core.batch.BatchRequest(
            datasource_name="pandas_datasource",
            data_connector_name="default_runtime_data_connector",
            data_asset_name="nyc_taxi_batch"
        ),
        expectation_suite=suite
    )
    validator.expect_column_values_to_not_be_null("tpep_pickup_datetime")
    validator.expect_column_values_to_be_between("trip_distance", min_value=0.1, max_value=100)
    validator.expect_column_values_to_be_between("fare_amount", min_value=0, max_value=500)
    results = validator.validate()
    return results.success

def convert_csv_to_parquet(csv_path: str, parquet_path: str) -> None:
    try:
        # Read CSV in chunks to handle large files
        chunk_iter = pd.read_csv(csv_path, chunksize=100000)
        for i, chunk in enumerate(chunk_iter):
            # Apply schema enforcement
            chunk["tpep_pickup_datetime"] = pd.to_datetime(chunk["tpep_pickup_datetime"])
            chunk["trip_distance"] = chunk["trip_distance"].astype(float)
            # Write first chunk with schema, append subsequent chunks
            if i == 0:
                chunk.to_parquet(
                    parquet_path,
                    engine="pyarrow",
                    compression="snappy",
                    index=False
                )
            else:
                chunk.to_parquet(
                    parquet_path,
                    engine="pyarrow",
                    compression="snappy",
                    index=False,
                    append=True
                )
        logger.info(f"Converted {csv_path} to {parquet_path}")
    except Exception as e:
        logger.error(f"CSV to Parquet conversion failed: {e}")
        raise

// Define incremental refresh range: last 30 days, keep 365 days history
Incremental Refresh Start = 
    IF(
        MAX('nyc_taxi_2026'[tpep_pickup_datetime]) > TODAY() - 365,
        MAX('nyc_taxi_2026'[tpep_pickup_datetime]) - 365,
        BLANK()
    )

Incremental Refresh End = 
    IF(
        MAX('nyc_taxi_2026'[tpep_pickup_datetime]) < TODAY(),
        MAX('nyc_taxi_2026'[tpep_pickup_datetime]),
        TODAY()
    )

real-datasets-step-by-step/
├── data/
│   ├── raw/                # Downloaded monthly taxi Parquet files
│   └── clean/              # Cleaned, merged Parquet dataset
├── scripts/
│   ├── ingest_dataset.py   # Python ingestion script (Step 1)
│   └── deploy_powerbi.py   # Power BI API automation script (Step 3)
├── dax/
│   └── data_model.dax      # DAX data model definition (Step 2)
├── powerbi/
│   └── nyc_taxi_2026.pbix  # Pre-built Power BI report template
├── tests/
│   └── test_ingestion.py   # Unit tests for ingestion pipeline
└── README.md               # Tutorial setup instructions