---
name: sas-to-databricks-migration
description: Convert SAS analytical code (PROC SQL, DATA steps, PROC MEANS, PROC FREQ) to PySpark or Spark SQL for healthcare payer analytics. Handles HEDIS measures, risk adjustment (RAF/HCC), claims analytics, and prior authorization workflows.
---

# SAS to Databricks Migration Skill

Convert legacy SAS analytical code to production-ready PySpark or Spark SQL, optimized for healthcare payer analytics on Databricks.

## When to Use This Skill

Use this skill when:
- Converting SAS programs to PySpark or Spark SQL
- Migrating healthcare payer analytics (HEDIS, RAF, claims processing)
- Translating PROC SQL, PROC MEANS, PROC FREQ, or DATA steps
- Working with Medicare Advantage, Medicaid, or commercial health plan data

## Environment & Naming Conventions

### Catalogs & Schemas
- **Source**: `payer_dev.analytics_gold` (existing gold tables)
- **Target**: `payer_analyst_dev.<workflow_schema>` (converted outputs)
  - Schemas: `hedis_reports`, `risk_adjustment`, `claims_analytics`, `provider_analytics`, `member_analytics`, `prior_auth_analytics`
- Always use three-level namespace: `catalog.schema.table`

### Data Conventions
- **Date columns**: Use DATE type, ISO format `'2023-01-01'` (not SAS `'01JAN2023'd`)
- **Member IDs**: `member_id` (STRING)
- **Provider IDs**: `provider_npi` (STRING, 10 digits)
- **Diagnosis codes**: `icd10_code` (STRING with decimal, e.g., 'E11.9')
- **Procedure codes**: `cpt_code` or `hcpcs_code` (STRING, 5 characters)
- Always filter by `service_date` when working with claims/measures

## Output Format Decision

### When User Selects "PySpark":
- Use PySpark DataFrame API throughout
- Always import: `from pyspark.sql import functions as F`
- Use: `df.withColumn()`, `F.when()`, `df.filter()`, `df.groupBy()`

### When User Selects "SQL":
- Use **Hybrid approach**: `spark.sql()` for safe queries, PySpark for risky operations
- **SAFE for spark.sql()**: Simple SELECT, WHERE filters, aggregations without new columns
- **RISKY - Use df.withColumn()**: SELECT *, adding derived columns, column transformations
- Always import both: `from pyspark.sql import functions as F` and `from pyspark.sql import Window`

## Common SAS to PySpark/SQL Patterns

### PROC SQL → PySpark/SQL

**SAS Example**:
```sas
PROC SQL;
    SELECT member_id, plan_type, raf_score
    FROM gold.members
    WHERE measurement_year = 2024 AND active_flag = 1;
QUIT;
```

**PySpark**:
```python
from pyspark.sql import functions as F
members_df = spark.table("payer_dev.analytics_gold.members")
result_df = members_df.filter(
    (F.col("measurement_year") == 2024) & (F.col("active_flag") == 1)
).select("member_id", "plan_type", "raf_score")
```

**Spark SQL**:
```python
result = spark.sql("""
    SELECT member_id, plan_type, raf_score
    FROM payer_dev.analytics_gold.members
    WHERE measurement_year = 2024 AND active_flag = 1
""")
```

### DATA Step IF-THEN-ELSE → PySpark F.when()

**SAS Example**:
```sas
DATA work.member_categories;
    SET gold.members;
    IF age < 18 THEN age_group = 'Pediatric';
    ELSE IF age >= 18 AND age < 65 THEN age_group = 'Adult';
    ELSE age_group = 'Senior';
RUN;
```

**PySpark**:
```python
members_df = spark.table("payer_dev.analytics_gold.members")
result_df = members_df.withColumn(
    "age_group",
    F.when(F.col("age") < 18, "Pediatric")
     .when((F.col("age") >= 18) & (F.col("age") < 65), "Adult")
     .otherwise("Senior")
)
```

### PROC MEANS → PySpark groupBy().agg()

**SAS Example**:
```sas
PROC MEANS DATA=gold.claims NOPRINT;
    CLASS member_id;
    VAR paid_amount;
    OUTPUT OUT=claim_summary
        SUM(paid_amount)=total_paid
        MEAN(paid_amount)=avg_paid
        N(*)=claim_count;
RUN;
```

**PySpark**:
```python
claims_df = spark.table("payer_dev.analytics_gold.claims")
claim_summary_df = claims_df.groupBy("member_id").agg(
    F.sum("paid_amount").alias("total_paid"),
    F.mean("paid_amount").alias("avg_paid"),
    F.count("*").alias("claim_count")
)
claim_summary_df.write.mode("overwrite").saveAsTable(
    "payer_analyst_dev.claims_analytics.claim_summary"
)
```

### RETAIN/LAG → Window Functions

**SAS Example**:
```sas
DATA work.member_changes;
    SET gold.member_history;
    BY member_id;
    RETAIN prior_plan_type;
    IF FIRST.member_id THEN prior_plan_type = '';
    plan_changed = (plan_type NE prior_plan_type);
    prior_plan_type = plan_type;
RUN;
```

**PySpark**:
```python
from pyspark.sql.window import Window

member_history_df = spark.table("payer_dev.analytics_gold.member_history")
window_spec = Window.partitionBy("member_id").orderBy("effective_date")

result_df = member_history_df.withColumn(
    "prior_plan_type",
    F.lag("plan_type", 1).over(window_spec)
).withColumn(
    "plan_changed",
    F.when(
        (F.col("plan_type") != F.col("prior_plan_type")) & 
        F.col("prior_plan_type").isNotNull(),
        True
    ).otherwise(False)
)
```

## Function Mappings (Quick Reference)

### Date Functions
| SAS | PySpark | SQL |
|-----|---------|-----|
| `TODAY()` | `F.current_date()` | `CURRENT_DATE()` |
| `YEAR(date)` | `F.year(col)` | `YEAR(col)` |
| `INTCK('day',start,end)` | `F.datediff(end,start)` | `DATEDIFF(end,start)` |
| `INTNX('month',date,n)` | `F.add_months(col,n)` | `ADD_MONTHS(col,n)` |

### String Functions
| SAS | PySpark | SQL |
|-----|---------|-----|
| `UPCASE(str)` | `F.upper(col)` | `UPPER(col)` |
| `SUBSTR(str,pos,len)` | `F.substring(col,pos,len)` | `SUBSTRING(col,pos,len)` |
| `TRIM(str)` | `F.trim(col)` | `TRIM(col)` |
| `INDEX(str,substr)` | `F.instr(col,substr)` | `INSTR(col,substr)` |

### Aggregation Functions
| SAS | PySpark | SQL |
|-----|---------|-----|
| `SUM()` | `F.sum(col)` | `SUM(col)` |
| `MEAN()` | `F.mean(col)` or `F.avg(col)` | `AVG(col)` |
| `COUNT()` | `F.count(col)` | `COUNT(col)` |
| `MIN()` / `MAX()` | `F.min(col)` / `F.max(col)` | `MIN(col)` / `MAX(col)` |

## Critical Patterns to Remember

### 1. NULL Handling
SAS treats missing values differently than Spark. Always handle NULLs explicitly:

```python
# Use coalesce for NULL defaults
df = df.withColumn("total", 
    F.coalesce(F.col("value1"), F.lit(0)) + 
    F.coalesce(F.col("value2"), F.lit(0))
)
```

### 2. Type Casting
SAS auto-converts types; Spark requires explicit casting:

```python
# Always cast IDs to STRING for joins
result = members_df.alias("m").join(
    claims_df.alias("c"),
    F.col("m.member_id").cast("string") == F.col("c.member_id").cast("string"),
    "left"
)

# Cast aggregation results immediately
.agg(
    F.count("*").cast("long").alias("total_count"),
    F.sum("amount").cast("double").alias("total_amount"),
    F.avg("days").cast("double").alias("avg_days")
)
```

### 3. Case Sensitivity
SAS is case-insensitive, PySpark is case-sensitive by default:

```python
# ❌ WRONG - Inconsistent casing causes errors
df.filter(F.col("MEMBER_ID") == "12345")  # Error if column is "member_id"

# ✅ CORRECT - Use consistent snake_case
df.filter(F.col("member_id") == "12345")

# Check schema first
df.printSchema()
# Or: spark.sql("DESCRIBE TABLE payer_dev.analytics_gold.members")
```

### 4. Explicit Ordering
SAS sorts implicitly; Spark DataFrames are unordered by default:

```python
# Always add .orderBy() if order matters
result_df = members_df.select("member_id", "plan_type", "raf_score") \
    .orderBy(F.col("raf_score").desc_nulls_last())
```

### 5. Division by Zero
Always check denominator before division:

```python
# PySpark
df = df.withColumn(
    "rate",
    F.when(F.col("total") > 0, (F.col("numerator") / F.col("total")) * 100)
     .otherwise(None)
)

# SQL
CASE 
    WHEN total > 0 THEN (numerator / total) * 100
    ELSE NULL
END AS rate
```

## Healthcare Payer-Specific Examples

### HEDIS Breast Cancer Screening (BCS)

**SAS**:
```sas
PROC SQL;
    CREATE TABLE work.bcs_compliant AS
    SELECT m.member_id, m.member_name, m.age,
           CASE WHEN COUNT(DISTINCT c.service_date) >= 1 
                THEN 'Compliant' ELSE 'Non-Compliant' END AS compliance_status
    FROM gold.members m
    LEFT JOIN gold.claims c ON m.member_id = c.member_id
        AND c.cpt_code IN ('77065','77066','77067')
        AND c.service_date BETWEEN '01JAN2023'd AND '31DEC2024'd
    WHERE m.gender = 'F' AND m.age BETWEEN 50 AND 74
    GROUP BY m.member_id, m.member_name, m.age;
QUIT;
```

**PySpark**:
```python
members_df = spark.table("payer_dev.analytics_gold.members")
claims_df = spark.table("payer_dev.analytics_gold.claims")

eligible_members_df = members_df.filter(
    (F.col("gender") == "F") & F.col("age").between(50, 74)
)

screening_claims_df = claims_df.filter(
    F.col("cpt_code").isin(['77065','77066','77067']) &
    (F.col("service_date") >= '2023-01-01') &
    (F.col("service_date") <= '2024-12-31')
).select("member_id", "service_date").distinct()

result_df = eligible_members_df.alias("m").join(
    screening_claims_df.alias("c"),
    on=F.col("m.member_id") == F.col("c.member_id"),
    how="left"
).groupBy(
    F.col("m.member_id"), F.col("m.member_name"), F.col("m.age")
).agg(
    F.countDistinct("c.service_date").alias("screening_count")
).withColumn(
    "compliance_status",
    F.when(F.col("screening_count") >= 1, "Compliant").otherwise("Non-Compliant")
)

result_df.write.mode("overwrite").saveAsTable(
    "payer_analyst_dev.hedis_reports.bcs_summary"
)
```

### RAF Score Calculation

**SAS**:
```sas
PROC SQL;
    CREATE TABLE work.member_raf AS
    SELECT m.member_id, m.age, m.gender,
           SUM(h.coefficient) AS raf_score,
           COUNT(DISTINCT h.hcc_code) AS hcc_count
    FROM gold.members m
    LEFT JOIN gold.member_hccs h ON m.member_id = h.member_id
        AND h.payment_year = 2024
    WHERE m.measurement_year = 2024
    GROUP BY m.member_id, m.age, m.gender;
QUIT;
```

**PySpark**:
```python
members_df = spark.table("payer_dev.analytics_gold.members")
member_hccs_df = spark.table("payer_dev.analytics_gold.member_hccs")

member_raf_df = members_df.filter(
    F.col("measurement_year") == 2024
).alias("m").join(
    member_hccs_df.filter(F.col("payment_year") == 2024).alias("h"),
    on=F.col("m.member_id") == F.col("h.member_id"),
    how="left"
).groupBy(
    F.col("m.member_id"), F.col("m.age"), F.col("m.gender")
).agg(
    F.sum("h.coefficient").alias("raf_score"),
    F.countDistinct("h.hcc_code").alias("hcc_count")
)

member_raf_df.write.mode("overwrite").saveAsTable(
    "payer_analyst_dev.risk_adjustment.member_raf_scores"
)
```

## Validation Pattern (Always Include)

After every conversion, add a data quality validation step:

```python
# ============================================================
# DATA QUALITY VALIDATION
# ============================================================

print("\n✅ DATA QUALITY CHECKS:")

# Check for NULL values
null_count = result_df.filter(F.col("critical_column").isNull()).count()
print(f"   NULL values in critical_column: {null_count}")

# Check for negative values (if applicable)
negative_count = result_df.filter(F.col("amount") < 0).count()
print(f"   Negative amounts: {negative_count}")

# Record count reconciliation
input_count = source_df.count()
output_count = result_df.count()
print(f"   Input records: {input_count:,}")
print(f"   Output records: {output_count:,}")
print(f"   Match: {'✅ YES' if input_count == output_count else '❌ NO'}")

# Statistical summary
print("\n📊 Statistical Summary:")
result_df.select(
    F.min("amount").alias("min"),
    F.max("amount").alias("max"),
    F.avg("amount").alias("avg")
).show()

# Critical assertions (fail-fast)
assert null_count == 0, "VALIDATION FAILED: NULL values in critical column"
assert negative_count == 0, "VALIDATION FAILED: Negative amounts found"
assert input_count == output_count, "VALIDATION FAILED: Record count mismatch"
```

## Performance Optimization Tips

### Broadcast Small Tables
```python
from pyspark.sql.functions import broadcast
result_df = large_df.join(broadcast(small_lookup_df), "lookup_key")
```

### Partition Output Tables
```python
df.write.mode("overwrite") \
    .partitionBy("measurement_year", "plan_type") \
    .saveAsTable("target_table")
```

### Cache Reused DataFrames
```python
members_df = spark.table("payer_dev.analytics_gold.members").cache()
# Use multiple times
result1 = members_df.filter(F.col("age") >= 65)
result2 = members_df.filter(F.col("age") < 65)
members_df.unpersist()  # Clean up
```

### Filter Early
```python
# Good: Filter before expensive operations
result_df = claims_df \
    .filter(F.col("measurement_year") == 2024) \
    .filter(F.col("claim_status") == 'PAID') \
    .groupBy("member_id").agg(F.sum("paid_amount"))
```

## Common Pitfalls & Solutions

### Pitfall 1: Python vs Column Expressions
Never mix Python boolean logic with Column expressions:

```python
# ❌ WRONG
completed_count = df.count()  # Python int
result = df.withColumn("rate", 
    F.when(completed_count > 0, ...)  # Python boolean in Column expression!
)

# ✅ CORRECT
if completed_count > 0:  # Python if-statement
    result = df.withColumn("rate", ...)
else:
    result = df.withColumn("rate", F.lit(None))
```

### Pitfall 2: SELECT * with Derived Columns (SQL)
Causes COLUMN_ALREADY_EXISTS errors:

```sql
-- ❌ WRONG
SELECT *, 
       DATEDIFF(approval_date, request_date) AS days
FROM table

-- ✅ CORRECT - Use temp view
CREATE OR REPLACE TEMP VIEW clean_data AS
SELECT *, 
       DATEDIFF(approval_date, request_date) AS days
FROM table;

SELECT * FROM clean_data WHERE days > 5;
```

### Pitfall 3: Date Format Confusion
```python
# SAS uses: '01JAN2024'd
# PySpark/SQL uses: '2024-01-01'

# Always use ISO format in PySpark/SQL
df.filter(F.col("service_date") >= '2024-01-01')
```

## Macro Variable Handling

**SAS**:
```sas
%LET measurement_year = 2024;
%LET min_age = 18;
```

**PySpark**:
```python
# Use Python variables with f-strings
measurement_year = 2024
min_age = 18

result_df = members_df.filter(
    (F.col("year") == measurement_year) & (F.col("age") >= min_age)
)

# Or in SQL
result = spark.sql(f"""
    SELECT * 
    FROM payer_dev.analytics_gold.members
    WHERE year = {measurement_year} AND age >= {min_age}
""")
```

## Coding Standards

### PySpark
1. Always import: `from pyspark.sql import functions as F`
2. Use `snake_case` for variables, functions, tables
3. Add `_df` suffix to DataFrames: `members_df`, `claims_df`
4. Always use three-level namespace: `catalog.schema.table`
5. Handle NULLs explicitly in joins and aggregations
6. Add comments explaining business logic

### Databricks SQL
1. Always use three-level namespace
2. Use explicit JOIN keywords (never comma-separated)
3. ISO date format: `'YYYY-MM-DD'`
4. Add comments for business logic
5. Consistent formatting (aligned SELECT columns)

## Additional Resources

See also:
- [Databricks PySpark DataFrame API](https://spark.apache.org/docs/latest/api/python/reference/pyspark.sql/dataframe.html)
- [Databricks SQL Language Reference](https://docs.databricks.com/sql/language-manual/index.html)
- [Delta Lake Documentation](https://docs.databricks.com/delta/index.html)
- [Unity Catalog Best Practices](https://docs.databricks.com/data-governance/unity-catalog/best-practices.html)