Fabric administrator’s perspective

I would enrol in this course even if you don't want to get certified, it will be able to answer most of your questions: Microsoft Certified: Fabric Data Engineer Associate - Certifications | Microsoft Learn

hi,

You can go through this training on Microsoft Learn ,by the end of it you will be able to understand and implement  the tenant settings, workspace governance, access control, backup and recovery, monitoring, auditing, capacity management, deployment practices, and business continuity.

Manage a Microsoft Fabric Environment - Training | Microsoft Learn

Hi reddyr2502​,

The very first thing I would recommend is setting up a process to collect audit logs. The API only stores them for 28 days, so if you want a full audit history of who did what, you need to start capturing them and storing them. 

Here is the PySpark script that I use for this. I schedule it daily but it is designed to catch up if up to 14 days are missed for whatever reason. 

 import requests import notebookutils import json from datetime import datetime, timedelta, timezone  import pyspark.sql.functions as F from pyspark.sql.types import StructType, StructField, StringType, TimestampType from delta.tables import DeltaTable  # --------------------------------------------------------------------------- # Config # ---------------------------------------------------------------------------  API_URL           = "https://api.powerbi.com/v1.0/myorg/admin/activityevents" AUDITLOG_TABLE    = "LH_Fabric_Meta.bronze.auditlog" VIEW_REPORT_TABLE = "LH_Fabric_Meta.bronze.auditlog_view_report"  # How far back to scan for gaps.  Must not exceed the API retention window. API_MAX_DAYS_BACK  = 28 # How many calendar days to check for gaps.  Keep <= API_MAX_DAYS_BACK. GAP_LOOKBACK_DAYS  = 14  VIEW_REPORT_JSON_SCHEMA = StructType([     StructField("UserId",        StringType(), True),     StructField("ItemName",      StringType(), True),     StructField("DatasetName",   StringType(), True),     StructField("ReportName",    StringType(), True),     StructField("WorkSpaceName", StringType(), True),     StructField("ReportId",      StringType(), True), ])  # --------------------------------------------------------------------------- # API helpers # ---------------------------------------------------------------------------  def _get_token():     return notebookutils.credentials.getToken("pbi")   def _day_window_iso(date: datetime.date):     """Return (start_iso, end_iso) for a full UTC day given a date object."""     start = datetime(date.year, date.month, date.day, 0, 0, 0, tzinfo=timezone.utc)     end   = datetime(date.year, date.month, date.day, 23, 59, 59, tzinfo=timezone.utc)     # Power BI API expects millisecond precision without offset suffix     return (         start.strftime("%Y-%m-%dT%H:%M:%S.000"),         end.strftime(  "%Y-%m-%dT%H:%M:%S.999"),     )   def query_powerbi_audit(token: str, start_iso: str, end_iso: str):     """     Fetch all activity events for the given UTC window.     Handles continuation URI pagination automatically.     Raises on HTTP errors (will surface as notebook failure).     """     headers = {         "Authorization": f"Bearer {token}",         "Content-Type":  "application/json",     }     params = {         "startDateTime": f"'{start_iso}'",         "endDateTime":   f"'{end_iso}'",     }      all_events  = []     current_url = API_URL      while True:         response = requests.get(current_url, headers=headers, params=params, timeout=60)         response.raise_for_status()         data = response.json()          all_events.extend(data.get("activityEventEntities", []))          if data.get("lastResultSet"):             break          current_url = data.get("continuationUri")         if not current_url:             break          params = {}  # continuation URI carries its own query string      return all_events   # --------------------------------------------------------------------------- # Gap detection # ---------------------------------------------------------------------------  def get_present_dates(table_name: str, lookback_days: int) -> set:     """     Return the set of UTC dates (as date objects) that already have rows     in the audit log within the lookback window.     Returns an empty set if the table does not exist.     """     if not spark.catalog.tableExists(table_name):         return set()      cutoff = datetime.now(timezone.utc).date() - timedelta(days=lookback_days)      rows = (         spark.table(table_name)         .filter(F.col("CreationTime") >= F.lit(cutoff.isoformat()))         .select(F.to_date("CreationTime").alias("log_date"))         .distinct()         .collect()     )     return {row["log_date"] for row in rows}   def resolve_dates_to_load(lookback_days: int, api_max_days_back: int) -> list:     """     Determine the sorted list of dates that need to be fetched:       1. Always include yesterday (UTC).       2. Scan the last `lookback_days` for dates missing from the audit log.       3. Exclude dates older than `api_max_days_back` (outside API retention).      Returns a sorted list of date objects (oldest first).     """     today     = datetime.now(timezone.utc).date()     yesterday = today - timedelta(days=1)      # Oldest date the API can still return     api_horizon = today - timedelta(days=api_max_days_back)      # Full calendar of expected dates within lookback window (excluding today —     # today's events are still accumulating so we never try to load them)     expected_dates = {         today - timedelta(days=d)         for d in range(1, lookback_days + 1)         if (today - timedelta(days=d)) >= api_horizon     }      present_dates = get_present_dates(AUDITLOG_TABLE, lookback_days)     gap_dates     = expected_dates - present_dates      # Union with yesterday (always reload to catch late-arriving events)     dates_to_load = gap_dates | {yesterday}      if not dates_to_load:         return []      return sorted(dates_to_load)  # oldest → newest   # --------------------------------------------------------------------------- # Write helpers # ---------------------------------------------------------------------------  def _auditlog_schema():     return StructType([         StructField("Id",           StringType(),    True),         StructField("CreationTime", StringType(),    True),         StructField("Operation",    StringType(),    True),         StructField("FullRecord",   StringType(),    True),     ])   def _build_auditlog_df(events: list):     rows = [         (             event.get("Id"),             event.get("CreationTime"),             event.get("Operation"),             json.dumps(event),         )         for event in events     ]     df = spark.createDataFrame(rows, schema=_auditlog_schema())     df = df.withColumn("CreationTime", F.to_timestamp("CreationTime"))     return df   def merge_into(df, table_name: str, merge_key: str = "Id"):     """     Idempotent merge by merge_key.     Creates the table on first load, then merges on subsequent runs.     """     if not spark.catalog.tableExists(table_name):         df.write.format("delta").mode("overwrite").saveAsTable(table_name)         print(f"  [INIT] {df.count()} rows written to {table_name}")         return      delta_table = DeltaTable.forName(spark, table_name)     (         delta_table.alias("target")         .merge(df.alias("source"), f"target.{merge_key} = source.{merge_key}")         .whenMatchedUpdateAll()         .whenNotMatchedInsertAll()         .execute()     )     print(f"  [MERGE] {df.count()} source rows merged into {table_name}")   # --------------------------------------------------------------------------- # View report silver transform # ---------------------------------------------------------------------------  def _transform_view_report(auditlog_df):     filtered = auditlog_df.filter(F.col("Operation") == "ViewReport")     parsed   = filtered.withColumn(         "FullRecord_parsed",         F.from_json(F.col("FullRecord"), VIEW_REPORT_JSON_SCHEMA),     )     return parsed.select(         F.col("Id"),         F.col("CreationTime"),         F.col("FullRecord_parsed.UserId").alias("UserId"),         F.col("FullRecord_parsed.DatasetName").alias("DatasetName"),         F.col("FullRecord_parsed.ReportName").alias("ReportName"),         F.col("FullRecord_parsed.WorkSpaceName").alias("WorkSpaceName"),         F.col("FullRecord_parsed.ReportId").alias("ReportId"),     )   # --------------------------------------------------------------------------- # Per-date fetch + write # ---------------------------------------------------------------------------  def process_date(token: str, target_date) -> dict:     """     Fetch all audit events for `target_date` (a date object, UTC),     merge them into the auditlog bronze table, and propagate ViewReport     events to the view_report table.      Returns a summary dict for reporting.     """     start_iso, end_iso = _day_window_iso(target_date)     print(f"\n>>> Processing {target_date} ({start_iso} → {end_iso})")      events = query_powerbi_audit(token, start_iso, end_iso)      if not events:         print(f"  No events returned for {target_date} — skipping write")         return {"date": target_date, "events": 0, "status": "empty"}      # --- Auditlog bronze ---     auditlog_df = _build_auditlog_df(events)     merge_into(auditlog_df, AUDITLOG_TABLE, merge_key="Id")      # --- ViewReport silver ---     vr_df = _transform_view_report(auditlog_df)     if not vr_df.isEmpty():         merge_into(vr_df, VIEW_REPORT_TABLE, merge_key="Id")     else:         print(f"  No ViewReport events for {target_date}")      return {"date": target_date, "events": len(events), "status": "ok"}   # --------------------------------------------------------------------------- # Main # ---------------------------------------------------------------------------  def main():     token = _get_token()      print("=" * 60)     print("Power BI Audit Log — Bronze Ingestion with Gap Detection")     print(f"Lookback window : {GAP_LOOKBACK_DAYS} days")     print(f"API horizon     : {API_MAX_DAYS_BACK} days")     print("=" * 60)      dates_to_load = resolve_dates_to_load(GAP_LOOKBACK_DAYS, API_MAX_DAYS_BACK)      if not dates_to_load:         print("All dates within lookback window are present. Nothing to load.")         return      present_dates = get_present_dates(AUDITLOG_TABLE, GAP_LOOKBACK_DAYS)     today         = datetime.now(timezone.utc).date()     yesterday     = today - timedelta(days=1)      print(f"\nDates to load ({len(dates_to_load)} total):")     for d in dates_to_load:         reason = []         if d == yesterday:             reason.append("yesterday")         if d not in present_dates:             reason.append("gap")         print(f"  {d}  [{', '.join(reason)}]")      results = []     for target_date in dates_to_load:         try:             result = process_date(token, target_date)             results.append(result)         except Exception as e:             print(f"  [ERROR] Failed to process {target_date}: {e}")             results.append({"date": target_date, "events": 0, "status": f"error: {e}"})      # --- Summary ---     print("\n" + "=" * 60)     print("Run Summary")     print("=" * 60)     for r in results:         print(f"  {r['date']}  |  {r['events']:>6} events  |  {r['status']}")      errors = [r for r in results if r["status"].startswith("error")]     if errors:         raise RuntimeError(             f"{len(errors)} date(s) failed to load: "             + ", ".join(str(r['date']) for r in errors)         )   main()