Retail Analytics Reference Architecture¶

# Azure ML training pipeline
from azureml.core import Workspace, Dataset, Experiment
from azureml.train.automl import AutoMLConfig
from azureml.core.compute import ComputeTarget

# Connect to workspace
ws = Workspace.from_config()

# Load training data from Synapse
dataset = Dataset.Tabular.from_sql_query(
    query="""
        SELECT
            product_id,
            store_id,
            sale_date,
            quantity_sold,
            price,
            promotion_flag,
            day_of_week,
            is_holiday,
            temperature,
            precipitation
        FROM gold.sales_daily
        WHERE sale_date >= DATEADD(year, -2, GETDATE())
    """,
    query_timeout=600,
    workspace=ws
)

# Configure AutoML
automl_config = AutoMLConfig(
    task='forecasting',
    primary_metric='normalized_root_mean_squared_error',
    experiment_timeout_minutes=60,
    training_data=dataset,
    label_column_name='quantity_sold',
    n_cross_validations=5,
    enable_early_stopping=True,
    forecasting_parameters={
        'time_column_name': 'sale_date',
        'forecast_horizon': 30,
        'target_lags': [1, 7, 14, 28],
        'target_rolling_window_size': 7,
        'freq': 'D',
        'group_columns': ['product_id', 'store_id']
    },
    compute_target='ml-compute-cluster'
)

# Run experiment
experiment = Experiment(ws, 'demand-forecasting')
run = experiment.submit(automl_config, show_output=True)
best_run, fitted_model = run.get_output()

# Deploy model
from azureml.core.model import Model, InferenceConfig
from azureml.core.webservice import AciWebservice

model = best_run.register_model(
    model_name='demand-forecast-model',
    model_path='outputs/model.pkl'
)

inference_config = InferenceConfig(
    entry_script='score.py',
    environment=best_run.get_environment()
)

deployment_config = AciWebservice.deploy_configuration(
    cpu_cores=2,
    memory_gb=4,
    auth_enabled=True,
    enable_app_insights=True
)

service = Model.deploy(
    workspace=ws,
    name='demand-forecast-service',
    models=[model],
    inference_config=inference_config,
    deployment_config=deployment_config
)

# Databricks notebook for association rules
from pyspark.sql.functions import *
from pyspark.ml.fpm import FPGrowth

# Prepare transaction baskets
baskets = (spark.table("silver.transaction_lines")
    .join(spark.table("silver.products"), "product_id")
    .groupBy("transaction_id")
    .agg(collect_set("product_id").alias("items"))
    .where(size("items") > 1)  # Multi-item baskets only
)

# FP-Growth algorithm
fpGrowth = FPGrowth(
    itemsCol="items",
    minSupport=0.01,  # 1% of transactions
    minConfidence=0.3  # 30% confidence
)

model = fpGrowth.fit(baskets)

# Frequent itemsets
frequent_itemsets = model.freqItemsets
frequent_itemsets.write.mode("overwrite").saveAsTable("gold.frequent_itemsets")

# Association rules
rules = model.associationRules
rules.write.mode("overwrite").saveAsTable("gold.association_rules")

# Example: Products to recommend with milk
milk_recommendations = (rules
    .where(array_contains("antecedent", "PROD_MILK_001"))
    .orderBy(desc("lift"))
    .select(
        "consequent",
        "confidence",
        "lift",
        "support"
    )
    .limit(10)
)

milk_recommendations.show()

# Price elasticity calculation
price_elasticity = spark.sql("""
    WITH price_changes AS (
        SELECT
            product_id,
            sale_date,
            price,
            quantity_sold,
            LAG(price) OVER (PARTITION BY product_id ORDER BY sale_date) as prev_price,
            LAG(quantity_sold) OVER (PARTITION BY product_id ORDER BY sale_date) as prev_quantity
        FROM gold.sales_daily
        WHERE promotion_flag = false
    )
    SELECT
        product_id,
        AVG(
            ((quantity_sold - prev_quantity) / prev_quantity) /
            ((price - prev_price) / prev_price)
        ) as price_elasticity,
        STDDEV(
            ((quantity_sold - prev_quantity) / prev_quantity) /
            ((price - prev_price) / prev_price)
        ) as elasticity_stddev,
        COUNT(*) as price_change_events
    FROM price_changes
    WHERE prev_price IS NOT NULL
        AND prev_quantity > 0
        AND ABS((price - prev_price) / prev_price) > 0.01  -- >1% price change
    GROUP BY product_id
    HAVING COUNT(*) >= 10  -- Sufficient data points
""")

# Optimal price recommendation
optimal_pricing = spark.sql("""
    WITH current_metrics AS (
        SELECT
            p.product_id,
            p.current_price,
            p.cost,
            e.price_elasticity,
            AVG(s.quantity_sold) as avg_daily_quantity,
            MAX(c.competitor_price) as competitor_price
        FROM silver.products p
        JOIN price_elasticity e ON p.product_id = e.product_id
        JOIN gold.sales_daily s ON p.product_id = s.product_id
        LEFT JOIN external.competitor_prices c ON p.sku = c.sku
        WHERE s.sale_date >= DATEADD(day, -30, GETDATE())
        GROUP BY p.product_id, p.current_price, p.cost, e.price_elasticity
    )
    SELECT
        product_id,
        current_price,
        competitor_price,
        CASE
            WHEN price_elasticity < -1.5 THEN current_price * 0.95  -- Elastic, reduce
            WHEN price_elasticity > -0.5 THEN current_price * 1.05  -- Inelastic, increase
            ELSE current_price  -- Keep current
        END as recommended_price,
        (recommended_price - cost) * avg_daily_quantity * 365 as projected_annual_margin
    FROM current_metrics
    WHERE recommended_price > cost * 1.1  -- Maintain minimum margin
""")

# Optimized Spark configuration for retail workloads
spark.conf.set("spark.sql.adaptive.enabled", "true")
spark.conf.set("spark.sql.adaptive.coalescePartitions.enabled", "true")
spark.conf.set("spark.sql.adaptive.skewJoin.enabled", "true")

# Delta Lake optimizations
spark.conf.set("spark.databricks.delta.properties.defaults.autoOptimize.optimizeWrite", "true")
spark.conf.set("spark.databricks.delta.properties.defaults.autoOptimize.autoCompact", "true")
spark.conf.set("spark.databricks.delta.optimizeWrite.enabled", "true")
spark.conf.set("spark.databricks.delta.autoCompact.enabled", "true")

# Partition pruning
spark.conf.set("spark.sql.sources.partitionOverwriteMode", "dynamic")

# Z-ordering for transaction table
spark.sql("""
    OPTIMIZE silver.transactions
    ZORDER BY (transaction_date, store_id, customer_id)
""")

# Vacuum old versions
spark.sql("VACUUM silver.transactions RETAIN 168 HOURS")  # 7 days

# Microsoft Purview sensitivity labels
from azure.purview.catalog import PurviewCatalogClient
from azure.identity import DefaultAzureCredential

credential = DefaultAzureCredential()
client = PurviewCatalogClient(
    endpoint="https://<purview-account>.purview.azure.com",
    credential=credential
)

# Apply labels to datasets
sensitivity_labels = {
    "silver.customers": "Confidential - PII",
    "silver.transactions": "Internal - Financial",
    "silver.products": "General",
    "gold.customer_360": "Highly Confidential - Customer Data"
}

for table, label in sensitivity_labels.items():
    client.entity.create_or_update(
        entity={
            "typeName": "azure_datalake_gen2_path",
            "attributes": {
                "qualifiedName": f"abfss://data@storage.dfs.core.windows.net/{table}",
                "name": table,
                "classifications": [{
                    "typeName": label.replace(" - ", "_")
                }]
            }
        }
    )

# Payment data tokenization
from azure.keyvault.secrets import SecretClient
from hashlib import sha256
import uuid

def tokenize_card_number(card_number, key_vault_client):
    """Replace card number with token, store mapping in Key Vault"""

    # Generate token
    token = str(uuid.uuid4())

    # Hash for verification
    card_hash = sha256(card_number.encode()).hexdigest()

    # Store in Key Vault (encrypted at rest)
    key_vault_client.set_secret(
        name=f"card-token-{token}",
        value=card_number,
        tags={"type": "payment_token", "hash": card_hash}
    )

    # Return token and last 4 digits for display
    return {
        "token": token,
        "last_four": card_number[-4:],
        "card_type": detect_card_type(card_number)
    }

# Store only tokens in data lake
transaction_data = {
    "transaction_id": "TXN_12345",
    "customer_id": "CUST_67890",
    "payment_token": "<TOKENIZED-PAYMENT-VALUE>",
    "last_four": "4532",
    "amount": 149.99
}

# Azure role assignments
from azure.mgmt.authorization import AuthorizationManagementClient
from azure.identity import DefaultAzureCredential

credential = DefaultAzureCredential()
auth_client = AuthorizationManagementClient(credential, subscription_id)

# Data Engineer role
role_assignments = {
    "data-engineers": {
        "role": "Storage Blob Data Contributor",
        "scope": "/subscriptions/{sub}/resourceGroups/{rg}/providers/Microsoft.Storage/storageAccounts/{storage}",
        "paths": ["/bronze", "/silver"]
    },
    "data-analysts": {
        "role": "Storage Blob Data Reader",
        "scope": "/subscriptions/{sub}/resourceGroups/{rg}/providers/Microsoft.Storage/storageAccounts/{storage}",
        "paths": ["/gold"]
    },
    "ml-engineers": {
        "role": "Storage Blob Data Contributor",
        "scope": "/subscriptions/{sub}/resourceGroups/{rg}/providers/Microsoft.Storage/storageAccounts/{storage}",
        "paths": ["/gold", "/ml-features"]
    }
}

# 1. Auto-pause Spark pools when idle
spark_config = {
    "auto_pause": {
        "enabled": True,
        "delay_in_minutes": 15
    },
    "auto_scale": {
        "enabled": True,
        "min_nodes": 3,
        "max_nodes": 10
    }
}

# 2. Use spot instances for non-critical workloads
databricks_cluster = {
    "node_type_id": "Standard_D16s_v3",
    "num_workers": 5,
    "azure_attributes": {
        "availability": "SPOT_WITH_FALLBACK_AZURE",
        "spot_bid_max_price": -1  # Pay up to on-demand price
    }
}

# 3. Partition pruning to reduce data scanning
sales_query = """
    SELECT *
    FROM gold.sales_daily
    WHERE year = 2024
        AND month = 1
        AND store_id IN ('STORE_001', 'STORE_002')
"""
# Scans only 2 stores in January 2024, not entire dataset

# 4. Serverless SQL for infrequent queries
# Pay only for data processed, not compute time

# 5. Data lifecycle management
lifecycle_policy = {
    "rules": [
        {
            "name": "move-to-cool",
            "type": "Lifecycle",
            "definition": {
                "filters": {
                    "blobTypes": ["blockBlob"],
                    "prefixMatch": ["bronze/"]
                },
                "actions": {
                    "baseBlob": {
                        "tierToCool": {"daysAfterModificationGreaterThan": 30},
                        "tierToArchive": {"daysAfterModificationGreaterThan": 180}
                    }
                }
            }
        }
    ]
}

# Azure Monitor metrics for retail analytics
from azure.monitor.query import MetricsQueryClient, MetricAggregationType

metrics_to_track = {
    "event_hubs": [
        "IncomingMessages",
        "OutgoingMessages",
        "ThrottledRequests",
        "ActiveConnections"
    ],
    "synapse_spark": [
        "SparkJobsRunning",
        "SparkJobsSucceeded",
        "SparkJobsFailed",
        "SparkPoolCPUUtilization"
    ],
    "cosmos_db": [
        "TotalRequests",
        "TotalRequestUnits",
        "ServerSideLatency",
        "AvailabilityPercentage"
    ],
    "data_factory": [
        "PipelineSucceededRuns",
        "PipelineFailedRuns",
        "ActivitySucceededRuns",
        "TriggerSucceededRuns"
    ]
}

# Alert rules
alert_rules = {
    "high_latency": {
        "metric": "ServerSideLatency",
        "threshold": 100,  # ms
        "operator": "GreaterThan",
        "severity": "Warning"
    },
    "low_availability": {
        "metric": "AvailabilityPercentage",
        "threshold": 99.9,
        "operator": "LessThan",
        "severity": "Critical"
    },
    "pipeline_failures": {
        "metric": "PipelineFailedRuns",
        "threshold": 3,
        "operator": "GreaterThan",
        "window": "PT15M",
        "severity": "Error"
    }
}

from applicationinsights import TelemetryClient
from applicationinsights.exceptions import enable

# Initialize client
tc = TelemetryClient(instrumentation_key='<key>')
enable(instrumentation_key='<key>')

# Track custom events
tc.track_event('InventoryAlert', {
    'product_id': 'PROD_12345',
    'store_id': 'STORE_001',
    'current_quantity': 5,
    'reorder_point': 10,
    'alert_type': 'Low Stock'
})

# Track custom metrics
tc.track_metric('DailyRevenue', 45623.50, properties={
    'store_id': 'STORE_001',
    'date': '2024-01-15'
})

# Track dependencies
tc.track_dependency('CosmosDB', 'GetRecommendations',
    duration=45, success=True)

tc.flush()