Python and R Foundations for Federal Platforms

# Databricks authentication — use environment variables, never hardcoded tokens
import os
from databricks.sdk import WorkspaceClient

# PAT injected at cluster launch via Databricks Secrets or environment config
# NEVER: client = WorkspaceClient(token="dapi1234567890abcdef")
client = WorkspaceClient(
    host=os.environ["DATABRICKS_HOST"],
    token=os.environ["DATABRICKS_TOKEN"],
)

me = client.current_user.me()
print(f"Authenticated as: {me.user_name}")

# Palantir Foundry: token is injected automatically inside the platform
# For local development:
import os
from foundry import FoundryClient

client = FoundryClient(
    auth_token=os.environ.get("FOUNDRY_TOKEN"),
    hostname=os.environ.get("FOUNDRY_HOSTNAME", "your-agency.palantirfoundry.com"),
)

import pandas as pd

# USASpending procurement data — always load NAICS as string, not int
df = pd.read_csv(
    "usaspending_fy2024_awards.csv",
    dtype={
        "contract_award_unique_key": str,
        "recipient_uei": str,
        "naics_code": str,           # "031" != 31 — always string
        "federal_action_obligation": float,
    },
    parse_dates=["action_date", "period_of_performance_start_date"],
    low_memory=False,
)

# Federal FY: Oct 1 to Sep 30
# FY2024 = Oct 1, 2023 through Sep 30, 2024
df["fiscal_year"] = df["action_date"].apply(
    lambda d: d.year if d.month < 10 else d.year + 1
)

print(f"Rows: {len(df):,}  |  Memory: {df.memory_usage(deep=True).sum() / 1e6:.1f} MB")

from pyspark.sql import functions as F

# In a Databricks notebook, spark is already available
df = spark.table("procurement_catalog.silver.usaspending_fy2024")

# PySpark operations are lazy — nothing runs until an action is called
df_filtered = (
    df
    .filter(F.col("federal_action_obligation") > 1_000_000)
    .filter(F.col("naics_sector_code").isin(["54", "51", "33"]))
    .withColumn("obligation_millions",
                F.round(F.col("federal_action_obligation") / 1_000_000, 2))
    .select(
        "contract_award_unique_key", "recipient_name",
        "naics_code", "obligation_millions", "action_date", "fiscal_year"
    )
)

print(f"Filtered count: {df_filtered.count():,}")

import numpy as np

obligations = np.array(df["federal_action_obligation"].values, dtype=np.float64)

print(f"Total obligations: ${obligations.sum():,.2f}")
print(f"Median: ${np.median(obligations):,.2f}")
print(f"P90: ${np.percentile(obligations, 90):,.2f}")
print(f"P99: ${np.percentile(obligations, 99):,.2f}")

from databricks.sdk import WorkspaceClient

w = WorkspaceClient()

# List jobs, trigger runs, inspect clusters
jobs = list(w.jobs.list())
print(f"Found {len(jobs)} jobs in workspace")

run = w.jobs.run_now(job_id=12345)
print(f"Job triggered, run ID: {run.run_id}")

for cluster in w.clusters.list():
    print(f"  {cluster.cluster_name}: {cluster.state.value}")

from transforms.api import transform_df, Input, Output
import pandas as pd

@transform_df(
    Output("/analytics/silver/procurement_cleaned"),
    raw=Input("/data/raw/usaspending_fy2024"),
)
def clean_procurement(raw: pd.DataFrame) -> pd.DataFrame:
    """
    Foundry transform: input/output are automatically version-controlled.
    Runs in Foundry's managed Spark environment.
    """
    df = raw.copy()
    df["naics_code"] = (
        df["naics_code"]
        .astype(str).str.strip()
        .str.replace(r"[^0-9]", "", regex=True)
        .str.zfill(6)
    )
    df["federal_action_obligation"] = pd.to_numeric(
        df["federal_action_obligation"], errors="coerce"
    )
    return df

import pkg_resources

def check_package_availability(packages: list) -> dict:
    results = {}
    for pkg in packages:
        try:
            dist = pkg_resources.get_distribution(pkg)
            results[pkg] = dist.version
        except pkg_resources.DistributionNotFound:
            results[pkg] = None
    return results

needed = ["scikit-learn", "xgboost", "shap", "mlflow", "hyperopt"]
status = check_package_availability(needed)
for pkg, version in status.items():
    status_str = f"v{version}" if version else "NOT INSTALLED"
    print(f"  {pkg:<20} {status_str}")

library(sparklyr)
library(dplyr)

sc <- spark_connect(method = "databricks")

df_spark <- spark_read_table(sc, "procurement_catalog.silver.usaspending_fy2024")

# dplyr verbs translate to Spark operations via sparklyr
summary_df <- df_spark %>%
  filter(federal_action_obligation > 1e6) %>%
  group_by(naics_sector_code, fiscal_year) %>%
  summarise(
    total_obligations = sum(federal_action_obligation, na.rm = TRUE),
    award_count = n(),
    .groups = "drop"
  ) %>%
  arrange(desc(total_obligations)) %>%
  collect()  # bring result to local R dataframe

print(head(summary_df, 10))