How to Automate Fabric Notebooks Using Pipelines in Microsoft Fabric?

from notebookutils import mssparkutils
from concurrent.futures import ThreadPoolExecutor, as_completed
from pyspark.sql import SparkSession
from pyspark.sql.functions import col

# Initialize Spark session
spark = SparkSession.builder.appName("_Data_Refresh").getOrCreate()

# List of notebooks to execute
notebooks = [
    "LedgerEntity",  
    "NumberSequenceEntity",
    "DimensionSetEntity",
    "FinancialDimensionValueEntity",     
    "LedgerChartOfAccountsEntity", 
    "MainAccountEntity", 
    "DirPartyBaseEntity",
    "DirPersonBaseEntity",
    "DirPartyEntity",
    "LogisticsPostalAddressBaseEntity",  
    "OMLegalEntity",  
    "ExchangeRateEntity",   
    "InventProcurementLedgerPostingDefinitionEntity",
    "PurchPurchaseOrderHeaderV2Entity",
    "PurchPurchaseOrderLineV2Entity",
    "VendVendorV2Entity",
    "VendVendorBankAccountEntity",
    "VendTransEntity",
    "HcmEmployeeV2Entity",
    "GeneralJournalAccountEntryReportingEntity",
    "EcoResProductV2Entity",
    "EcoResReleasedProductV2Entity",
    "AssetGroupEntity",
    "AssetBookV2Entity",
    "AssetFixedAssetV2Entity",
    "AssetTransactionListing"
]

# Function to run a single notebook and return the record counts and timestamp
def run_notebook(notebook_name):
    try:
        # Run the notebook and get the result (expects a string in format "new,total,timestamp")
        result = mssparkutils.notebook.run(notebook_name, timeout_seconds=60*60)
        new_records, total_records, timestamp = result.split(",")  # Split the values
        new_records, total_records = int(new_records), int(total_records)  # Convert counts to integers
        return (notebook_name, new_records, total_records, timestamp)
    except Exception as e:
        print(f"Notebook {notebook_name} failed: {str(e)}")
        return (notebook_name, -1, -1, None)  # -1 indicates failure, None for timestamp

# Function to execute multiple notebooks in parallel
def execute_notebooks(notebook_list, max_workers=5):
    row_counts = []
    with ThreadPoolExecutor(max_workers=max_workers) as executor:
        futures = {executor.submit(run_notebook, nb): nb for nb in notebook_list}
        for future in as_completed(futures):
            notebook_name, new_records, total_records, timestamp = future.result()

            if new_records == -1:
                raise Exception(f"Notebook {notebook_name} failed to execute.")
            else:
                print(f"Notebook {notebook_name} added {new_records} records. Total records: {total_records}. Timestamp: {timestamp}")

            # Append the row count data to the list
            row_counts.append((notebook_name, new_records, total_records, timestamp))

    # After processing all notebooks, write the results to the RowCount table
    write_row_counts_to_table(row_counts)

# Function to write row counts to a Microsoft Fabric Lakehouse Table
def write_row_counts_to_table(row_counts):
    # Create a DataFrame from the row_counts
    row_count_df = spark.createDataFrame(row_counts, ["NotebookName", "NewRecords", "TotalRecords", "ExecutionTimestamp"])

    # Perform a merge to ensure only one row per notebook
    target_table = "RowCount"

    # Use a temporary view for the source DataFrame to allow merging
    row_count_df.createOrReplaceTempView("source_row_counts")

    # Execute the merge (upsert) operation
    merge_query = f"""
    MERGE INTO {target_table} AS target
    USING source_row_counts AS source
    ON target.NotebookName = source.NotebookName
    WHEN MATCHED THEN
        UPDATE SET 
            target.NewRecords = source.NewRecords,
            target.TotalRecords = source.TotalRecords,
            target.ExecutionTimestamp = source.ExecutionTimestamp
    WHEN NOT MATCHED THEN
        INSERT (NotebookName, NewRecords, TotalRecords, ExecutionTimestamp)
        VALUES (source.NotebookName, source.NewRecords, source.TotalRecords, source.ExecutionTimestamp)
    """
    
    # Run the merge query
    spark.sql(merge_query)

# Execute the notebooks
try:
    execute_notebooks(notebooks)

except Exception as e:
    print(f"Error executing notebooks: {str(e)}")
    raise

<p>@{concat('RunID: ', string(pipeline().RunId), '; YourCompanyName ', pipeline().PipelineName, ' run has failed. ', 'Please check the pipeline run details in Fabric for more information. ', 'Pipeline Run Date: ', utcNow())}</p>

<p>@{concat('RunID: ', string(pipeline().RunId), '; YourCompanyName ', pipeline().PipelineName, ' run was successful. ', 'Please check the pipeline run details in Fabric for more information. ', 'Pipeline Run Date: ', utcNow())}</p>

# Save the result as a Delta table
result_df.write.format("delta").mode("overwrite").saveAsTable("AssetBookV2Entity")
print("The query result has been successfully saved as a table 'AssetBookV2Entity'.")

# Return row count
# Define your table name
table_name = "assetbookv2entity"

# Count the new records added in this execution
new_records_count = result_df.count()

# Get the total number of records in the table after insertion
total_records_count = spark.sql(f"SELECT COUNT(*) FROM {table_name}").collect()[0][0]

# Step 5: Optionally get the current timestamp for the execution
execution_timestamp = spark.sql("SELECT current_timestamp()").collect()[0][0]

# Return new and total records as a string for the main notebook
mssparkutils.notebook.exit(f"{new_records_count},{total_records_count},{execution_timestamp}")