Microsoft Staff-Level AI Engineer Interview Preparation Guide

from functools import lru_cache

def count_no4_up_to(N):
    s = str(N)
    L = len(s)

    @lru_cache(None)
    def dp(pos, tight, leading_zero):
        # pos: current index in s (0..L)
        # tight: 1 if prefix equals N's prefix else 0
        # leading_zero: 1 if all previous digits were zeros
        if pos == L:
            # Reached end: count this number.
            # We count 0 as valid (leading_zero True at end -> number 0)
            return 1
        limit = int(s[pos]) if tight else 9
        total = 0
        for d in range(0, limit + 1):
            if d == 4:
                continue  # forbid digit 4
            ntight = tight and (d == limit)
            nleading = leading_zero and (d == 0)
            total += dp(pos + 1, ntight, nleading)
        return total

    return dp(0, True, True)

# Examples
print(count_no4_up_to(0))      # 1 -> only 0
print(count_no4_up_to(9))      # 9 (0..9 except 4 => 9)
print(count_no4_up_to(100))    # counts numbers in [0,100] without digit 4

import time, math
# build balanced BST nodes from sorted keys
def build_balanced(keys):
    if not keys: return None
    mid = len(keys)//2
    left = build_balanced(keys[:mid])
    right = build_balanced(keys[mid+1:])
    return (keys[mid], left, right)

# flatten BFS order
def bfs_order(root):
    arr=[]; q=[root]
    while q:
        node=q.pop(0)
        if node is None: continue
        arr.append(node[0])
        q.append(node[1]); q.append(node[2])
    return arr

# vEB layout: recursively place root, then recursively left and right subtrees packed
def veb_order(root):
    if root is None: return []
    # compute sizes
    left, right = root[1], root[2]
    return [root[0]] + veb_order(left) + veb_order(right)

# simple indexed search on contiguous array assuming we stored tree structure (here we simulate lookups)
def linear_search_array(arr, target):
    # simulate tree search by binary-tree indices for balanced complete tree:
    i=0; n=len(arr)
    while i<n:
        if arr[i]==target: return True
        # simulate going left/right: left=2*i+1 right=2*i+2 if within bounds
        if target < arr[i]:
            i=2*i+1
        else:
            i=2*i+2
    return False

# microbenchmark
N=2**16
keys=list(range(N))
root=build_balanced(keys)
bfs=bfs_order(root)
veb=veb_order(root)
def bench(arr):
    t0=time.perf_counter()
    s=0
    for k in range(N):
        linear_search_array(arr, k)
    return time.perf_counter()-t0

print("BFS time", bench(bfs))
print("vEB time", bench(veb))

terraform {
  required_providers {
    azurerm = { source = "hashicorp/azurerm", version = ">=3.0" }
  }
}

provider "azurerm" {
  features {}
}

variable "rg_name"      { type = string }
variable "location"     { type = string }
variable "workspace_name"{ type = string }
variable "storage_name" { type = string }
variable "sp_object_id" { type = string } # service principal principal object id
variable "subscription_id" { type = string }

resource "azurerm_resource_group" "rg" {
  name     = var.rg_name
  location = var.location
}

resource "azurerm_storage_account" "sa" {
  name                     = var.storage_name
  resource_group_name      = azurerm_resource_group.rg.name
  location                 = azurerm_resource_group.rg.location
  account_tier             = "Standard"
  account_replication_type = "LRS"
}

resource "azurerm_machine_learning_workspace" "mlws" {
  name                = var.workspace_name
  location            = azurerm_resource_group.rg.location
  resource_group_name = azurerm_resource_group.rg.name

  storage_account_id = azurerm_storage_account.sa.id
  sku_name           = "Basic"
}

resource "azurerm_machine_learning_compute_cluster" "cluster" {
  name                         = "${var.workspace_name}-cluster"
  resource_group_name          = azurerm_resource_group.rg.name
  machine_learning_workspace_id = azurerm_machine_learning_workspace.mlws.id

  vm_size   = "Standard_D2_v2"
  min_node_count = 0
  max_node_count = 4

  # autoscale settings: enable autoscale
  autoscale {
    min_node_count = 0
    max_node_count = 4
  }

  # optional: scale settings and identity can be added per needs
}

resource "azurerm_role_assignment" "sp_ml_role" {
  scope                = azurerm_machine_learning_workspace.mlws.id
  role_definition_name = "Contributor" # or "Azure Machine Learning Workspace Contributor"
  principal_id         = var.sp_object_id
}

from pyspark.sql import functions as F
from pyspark.sql import Window

# assume spark session exists and transactions table is in a Hive/Delta table
transactions = spark.table("transactions")

# job run date (UTC or controlled timezone)
run_date = F.current_date()

# define lookback and grace days
lookback_days = 30
grace_days = 2

# filter by event time including grace to capture late arrivals
start_date = F.date_sub(run_date, lookback_days + grace_days - 1)

tx_filtered = (
    transactions
    .withColumn("occured_date", F.to_date("occured_at"))
    .filter(F.col("occured_date").between(start_date, run_date))
)

# optional: deduplicate by tx_id keeping latest occured_at (if source can duplicate)
tx_dedup = tx_filtered.withColumn(
    "row_num",
    F.row_number().over(Window.partitionBy("tx_id").orderBy(F.col("occured_at").desc()))
).filter(F.col("row_num") == 1).drop("row_num")

# compute aggregates using only the strict last 30 days window (exclude grace beyond 30)
tx_30d = tx_dedup.filter(F.col("occured_date").between(F.date_sub(run_date, lookback_days - 1), run_date))

result = (
    tx_30d
    .groupBy("user_id")
    .agg(
        F.count("*").alias("total_purchases_last_30_days"),
        F.avg("amount").alias("avg_purchase_amount_last_30_days")
    )
)

# write results (partition by run_date) for downstream use
result.withColumn("run_date", run_date) \
      .repartition(200, "user_id") \
      .write.mode("overwrite") \
      .partitionBy("run_date") \
      .format("parquet") \
      .save("/path/to/user_30d_aggregates/")

-- read recent partitions including grace
WITH tx_recent AS (
  SELECT tx_id, user_id, amount, occured_at, CAST(occured_at AS DATE) AS occured_date
  FROM transactions
  WHERE occured_date BETWEEN date_sub(current_date(), 31) AND current_date()
),
tx_dedup AS (
  SELECT * FROM (
    SELECT *, ROW_NUMBER() OVER (PARTITION BY tx_id ORDER BY occured_at DESC) rn
    FROM tx_recent
  ) WHERE rn = 1
),
tx_30d AS (
  SELECT * FROM tx_dedup
  WHERE occured_date BETWEEN date_sub(current_date(), 29) AND current_date()
)
SELECT user_id,
       COUNT(*) AS total_purchases_last_30_days,
       AVG(amount) AS avg_purchase_amount_last_30_days
FROM tx_30d
GROUP BY user_id;