Imported from bigdatavik/databricks-skills-legacy-migration (skills/sas-migration/SKILL.md). Install upstream with npx skills add bigdatavik/databricks-skills-legacy-migration --skill sas-migration. Copyright stays with the author.

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:

PROC SQL;
    SELECT member_id, plan_type, raf_score
    FROM gold.members
    WHERE measurement_year = 2024 AND active_flag = 1;
QUIT;

PySpark:

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:

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:

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:

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:

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:

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:

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:

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

String Functions

Aggregation Functions

Critical Patterns to Remember

1. NULL Handling

SAS treats missing values differently than Spark. Always handle NULLs explicitly:

# 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:

# 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:

# ❌ 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:

# 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:

# 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:

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:

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:

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:

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:

# ============================================================
# 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

from pyspark.sql.functions import broadcast
result_df = large_df.join(broadcast(small_lookup_df), "lookup_key")

Partition Output Tables

df.write.mode("overwrite") \
    .partitionBy("measurement_year", "plan_type") \
    .saveAsTable("target_table")

Cache Reused DataFrames

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

# 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:

# ❌ 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:

-- ❌ 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

# 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:

%LET measurement_year = 2024;
%LET min_age = 18;

PySpark:

# 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
• Databricks SQL Language Reference
• Delta Lake Documentation
• Unity Catalog Best Practices