Databricks

# Reading from Unity Catalog Delta table
df = spark.table("don_catalog.readiness.ship_maintenance_silver")

# Basic data quality check before analysis
from pyspark.sql import functions as F

quality_report = df.agg(
    F.count("*").alias("total_rows"),
    F.count_distinct("hull_number").alias("unique_ships"),
    F.sum(F.col("mission_capable_flag").cast("int")).alias("mc_count"),
    F.count(F.when(F.col("maintenance_date").isNull(), 1)).alias("null_dates"),
)
quality_report.show()

# Cleaning: standardize unit codes and dates
cleaned = (
    df
    .withColumn("unit_code", F.trim(F.upper(F.col("unit_code"))))
    .withColumn(
        "maintenance_date",
        F.to_date(F.col("maintenance_date"), "yyyy-MM-dd")
    )
    .dropDuplicates(["hull_number", "maintenance_date", "work_order_id"])
)

# Write cleaned data to silver tier
(
    cleaned.write
    .format("delta")
    .mode("overwrite")
    .option("mergeSchema", "true")
    .saveAsTable("don_catalog.readiness.ship_maintenance_silver_v2")
)

import mlflow
import mlflow.sklearn
from sklearn.ensemble import RandomForestClassifier
from sklearn.metrics import f1_score, classification_report
import pandas as pd

# Set experiment path (your email or team path in Databricks)
mlflow.set_experiment("/Users/your.email@agency.gov/maintenance-anomaly-v2")

with mlflow.start_run(run_name="rf_v3_balanced_classes"):
    # Log parameters
    params = {
        "n_estimators": 300,
        "max_depth": 10,
        "class_weight": "balanced",
        "random_state": 42,
    }
    mlflow.log_params(params)

    # Log data lineage (critical for audit)
    mlflow.log_param("training_table", "don_catalog.readiness.ship_maintenance_silver")
    mlflow.log_param("training_cutoff", "2025-01-01")
    mlflow.log_param("featurization_version", "v3.1")

    # Train
    clf = RandomForestClassifier(**params)
    clf.fit(X_train, y_train)
    y_pred = clf.predict(X_test)

    # Log metrics
    mlflow.log_metric("f1_macro", f1_score(y_test, y_pred, average="macro"))
    mlflow.log_metric("f1_weighted", f1_score(y_test, y_pred, average="weighted"))
    mlflow.log_metric("test_accuracy", clf.score(X_test, y_test))

    # Log the model with signature
    from mlflow.models import infer_signature
    signature = infer_signature(X_train, y_pred)
    mlflow.sklearn.log_model(
        clf,
        artifact_path="model",
        signature=signature,
        registered_model_name="maintenance-anomaly-detector",
    )

    # Log classification report as artifact
    report = classification_report(y_test, y_pred)
    mlflow.log_text(report, "classification_report.txt")

print(f"Run ID: {mlflow.active_run().info.run_id}")

-- Unity Catalog three-level namespace: catalog.schema.table
-- catalog = data domain or organization
-- schema  = data tier or subject area
-- table   = specific dataset

-- List catalogs you have access to
SHOW CATALOGS;

-- List schemas in a catalog
SHOW SCHEMAS IN don_catalog;

-- Examine table lineage and properties
DESCRIBE EXTENDED don_catalog.readiness.ship_maintenance_gold;

-- Check your permissions on a table
SHOW GRANTS ON TABLE don_catalog.readiness.ship_maintenance_gold;

-- Query with full namespace (always use three-part names)
SELECT hull_number, COUNT(*) AS maintenance_events
FROM don_catalog.readiness.ship_maintenance_gold
WHERE fiscal_year = 2025
  AND mission_capable_flag = TRUE
GROUP BY hull_number
ORDER BY maintenance_events DESC
LIMIT 20;