Tutorial: SAP Data to Microsoft Fabric¶

# Fabric Notebook: Transform SAP mirrored data to silver layer
# This notebook reads from the mirrored database and writes cleansed data

from pyspark.sql import SparkSession
from pyspark.sql import functions as F
from pyspark.sql.types import *

# Read mirrored SAP sales order data
df_vbak = spark.read.format("delta").load(
    "Tables/sap_mirror/vbak"
)

df_vbap = spark.read.format("delta").load(
    "Tables/sap_mirror/vbap"
)

# Transform: Silver layer - cleansed sales orders
silver_sales_orders = (
    df_vbak
    .filter(F.col("mandt") == "100")  # Filter to production client
    .select(
        F.col("vbeln").alias("sales_order_number"),
        F.col("erdat").alias("created_date"),
        F.col("erzet").alias("created_time"),
        F.col("ernam").alias("created_by"),
        F.col("auart").alias("order_type"),
        F.col("vkorg").alias("sales_org"),
        F.col("vtweg").alias("distribution_channel"),
        F.col("spart").alias("division"),
        F.col("kunnr").alias("customer_number"),
        F.col("netwr").alias("net_value"),
        F.col("waerk").alias("currency"),
        F.col("bstnk").alias("po_number"),
        F.col("lifsk").alias("delivery_block"),
        F.col("faksk").alias("billing_block"),
        F.to_date("erdat", "yyyyMMdd").alias("order_date")
    )
)

# Write silver layer
silver_sales_orders.write \
    .format("delta") \
    .mode("overwrite") \
    .option("overwriteSchema", "true") \
    .save("Tables/silver/sales_orders")

# Transform: Silver layer - sales order items
silver_sales_items = (
    df_vbap
    .filter(F.col("mandt") == "100")
    .select(
        F.col("vbeln").alias("sales_order_number"),
        F.col("posnr").alias("item_number"),
        F.col("matnr").alias("material_number"),
        F.col("arktx").alias("item_description"),
        F.col("kwmeng").alias("order_quantity"),
        F.col("vrkme").alias("sales_unit"),
        F.col("netwr").alias("net_value"),
        F.col("waerk").alias("currency"),
        F.col("werks").alias("plant"),
        F.col("lgort").alias("storage_location"),
        F.col("pstyv").alias("item_category")
    )
)

silver_sales_items.write \
    .format("delta") \
    .mode("overwrite") \
    .option("overwriteSchema", "true") \
    .save("Tables/silver/sales_order_items")

print("Silver layer transformation complete.")

# Gold layer: Sales analytics aggregations

# Read silver layer
df_orders = spark.read.format("delta").load("Tables/silver/sales_orders")
df_items = spark.read.format("delta").load("Tables/silver/sales_order_items")

# Gold: Daily sales summary
gold_daily_sales = (
    df_orders
    .groupBy("order_date", "sales_org", "order_type")
    .agg(
        F.count("*").alias("order_count"),
        F.sum("net_value").alias("total_net_value"),
        F.countDistinct("customer_number").alias("unique_customers")
    )
    .orderBy("order_date")
)

gold_daily_sales.write \
    .format("delta") \
    .mode("overwrite") \
    .save("Tables/gold/daily_sales_summary")

# Gold: Material sales performance
gold_material_sales = (
    df_items
    .join(df_orders, "sales_order_number")
    .groupBy("material_number", "item_description", "sales_org")
    .agg(
        F.sum("net_value").alias("total_revenue"),
        F.sum("order_quantity").alias("total_quantity"),
        F.count("*").alias("line_item_count")
    )
    .orderBy(F.desc("total_revenue"))
)

gold_material_sales.write \
    .format("delta") \
    .mode("overwrite") \
    .save("Tables/gold/material_sales_performance")

print("Gold layer aggregation complete.")

# Azure OpenAI: Natural language queries on SAP data
from openai import AzureOpenAI

client = AzureOpenAI(
    azure_endpoint="https://aoai-csa-inabox.openai.azure.com/",
    api_version="2024-02-15-preview"
)

# User asks a question about SAP data
user_question = "What were our top 5 selling materials last quarter?"

# Generate SQL from natural language
response = client.chat.completions.create(
    model="gpt-4o",
    messages=[{
        "role": "system",
        "content": """You are a data analyst. Generate Spark SQL queries against
        these tables:
        - gold.material_sales_performance (material_number, item_description,
          total_revenue, total_quantity, sales_org)
        - gold.daily_sales_summary (order_date, sales_org, order_type,
          order_count, total_net_value)
        Return only the SQL query, no explanation."""
    }, {
        "role": "user",
        "content": user_question
    }]
)

# Execute the generated SQL
sql_query = response.choices[0].message.content
result_df = spark.sql(sql_query)
result_df.show()

# Check mirroring status
mirror_status = spark.sql("""
    SELECT table_name,
           last_sync_timestamp,
           row_count,
           DATEDIFF(minute, last_sync_timestamp, current_timestamp()) AS minutes_behind
    FROM information_schema.mirror_status
    WHERE database_name = 'sap_mirror'
""")
mirror_status.show()