📂 OneLake Files vs Tables Decision Guide¶

# Land raw CSV files from external SFTP
raw_path = "Files/landing/casino/slot_telemetry/2026/04/27/"

# Read from landing zone
df_raw = spark.read.format("csv") \
    .option("header", "true") \
    .option("inferSchema", "true") \
    .load(f"abfss://{workspace}@onelake.dfs.fabric.microsoft.com/{lakehouse}/{raw_path}")

# After validation, promote to Tables (see promotion section)

# Store compliance document scans
doc_path = "Files/compliance/sar_filings/2026/04/"

# Store metadata about the documents in a Delta table
doc_metadata = spark.createDataFrame([
    ("SAR-2026-04-001", "Files/compliance/sar_filings/2026/04/sar_001.pdf", "2026-04-15", "pending_review"),
    ("SAR-2026-04-002", "Files/compliance/sar_filings/2026/04/sar_002.pdf", "2026-04-16", "submitted"),
], ["filing_id", "document_path", "filing_date", "status"])

doc_metadata.write.format("delta").mode("append").saveAsTable("gold.sar_filing_index")

# Files section supports any format
files_structure = """
Files/
├── landing/
│   ├── csv/          # Partner CSV drops
│   ├── json/         # API response snapshots
│   └── xml/          # Legacy system exports
├── models/
│   ├── onnx/         # ML model artifacts
│   └── checkpoints/  # Training checkpoints
├── compliance/
│   ├── sar_filings/  # PDF scans
│   └── ctr_filings/  # PDF scans
└── temp/
    └── spark_staging/ # Intermediate outputs
"""

# This table is automatically available to Direct Lake
df_gold.write.format("delta") \
    .mode("overwrite") \
    .option("overwriteSchema", "true") \
    .saveAsTable("gold.slot_performance_daily")

# V-Order is applied automatically, optimizing Direct Lake reads

from pyspark.sql.types import StructType, StructField, StringType, DoubleType, TimestampType

slot_schema = StructType([
    StructField("machine_id", StringType(), False),
    StructField("timestamp", TimestampType(), False),
    StructField("coin_in", DoubleType(), True),
    StructField("coin_out", DoubleType(), True),
    StructField("denomination", DoubleType(), False),
])

# Schema enforcement rejects malformed records
df_validated = spark.read.schema(slot_schema).json("Files/landing/slot_events/")
df_validated.write.format("delta").mode("append").saveAsTable("bronze.slot_telemetry")

# Time travel for compliance audits
df_as_of = spark.read.format("delta") \
    .option("timestampAsOf", "2026-04-15T00:00:00Z") \
    .table("silver.player_transactions")

# Restore a table to a prior version after bad data load
from delta.tables import DeltaTable
dt = DeltaTable.forName(spark, "silver.player_transactions")
dt.restoreToVersion(42)

# Using Lakehouse Schemas to organize by medallion layer
# Tables appear under: Tables/bronze/*, Tables/silver/*, Tables/gold/*

df.write.format("delta").mode("append").saveAsTable("bronze.slot_telemetry")
df.write.format("delta").mode("overwrite").saveAsTable("silver.slot_cleansed")
df.write.format("delta").mode("overwrite").saveAsTable("gold.slot_kpi_daily")

# SQL access with schema prefix
# SELECT * FROM lh_main.bronze.slot_telemetry

# Step 1: Read from Files landing zone
df_landing = spark.read.format("csv") \
    .option("header", "true") \
    .schema(defined_schema) \
    .load("Files/landing/usda_crop_production/2026/04/")

# Step 2: Validate
from pyspark.sql.functions import col, when, count

null_check = df_landing.select([
    count(when(col(c).isNull(), c)).alias(c) for c in df_landing.columns
])
null_counts = null_check.collect()[0]
for field in defined_schema.fields:
    if not field.nullable and null_counts[field.name] > 0:
        raise ValueError(f"Non-nullable field {field.name} has {null_counts[field.name]} nulls")

# Step 3: Write to Tables as Delta
df_landing.write.format("delta") \
    .mode("append") \
    .option("mergeSchema", "true") \
    .saveAsTable("bronze.usda_crop_production")

# Step 4: Archive processed files
mssparkutils.fs.mv(
    "Files/landing/usda_crop_production/2026/04/",
    "Files/archive/usda_crop_production/2026/04/"
)

from pyspark.sql.functions import max as spark_max

# Get high watermark from last successful load
watermark_df = spark.sql("SELECT max(ingestion_timestamp) as hwm FROM bronze.epa_air_quality")
high_watermark = watermark_df.collect()[0]["hwm"]

# Read only new files
df_new = spark.read.format("json") \
    .load("Files/landing/epa_air_quality/") \
    .filter(col("_metadata.file_modification_time") > high_watermark)

if df_new.count() > 0:
    df_new.withColumn("ingestion_timestamp", current_timestamp()) \
        .write.format("delta").mode("append") \
        .saveAsTable("bronze.epa_air_quality")

# ❌ WRONG: CSV files placed in Tables/ folder manually
# They won't be recognized as Delta tables and will cause errors
mssparkutils.fs.cp("Files/landing/data.csv", "Tables/data.csv")  # BROKEN

# ✅ CORRECT: Read CSV and write as Delta table
df = spark.read.csv("Files/landing/data.csv", header=True)
df.write.format("delta").saveAsTable("bronze.imported_data")

# ❌ WRONG: Storing image bytes in Delta tables
df_images = spark.createDataFrame([
    ("img_001", open("surveillance_001.jpg", "rb").read()),
], ["image_id", "image_bytes"])
df_images.write.format("delta").saveAsTable("bronze.surveillance")  # Huge, slow, wasteful

# ✅ CORRECT: Store path reference in table, file in Files
df_index = spark.createDataFrame([
    ("img_001", "Files/surveillance/2026/04/27/cam_01_001.jpg", "2026-04-27T14:30:00Z"),
], ["image_id", "file_path", "capture_time"])
df_index.write.format("delta").saveAsTable("bronze.surveillance_index")

# ❌ WRONG: Keeping Parquet in Files and querying with Spark repeatedly
df = spark.read.parquet("Files/analytics/player_summary/")  # No caching, no stats, no V-Order

# ✅ CORRECT: Promote to Tables for repeated queries
df.write.format("delta").saveAsTable("gold.player_summary")  # V-Order, data skipping, SQL access

# ❌ WRONG: Writing external data directly to Tables without validation
spark.read.csv("https://external-api.com/data.csv") \
    .write.format("delta").saveAsTable("bronze.external")  # No validation, no replay

# ✅ CORRECT: Land first, validate, then promote
# 1. Download to Files/landing/
# 2. Validate schema + quality
# 3. Write validated data to Tables

# Pattern: PII in Files (encrypted), metadata in Tables
# Scenario: Player social security numbers for W-2G compliance

# ❌ WRONG: SSN stored in a Delta table accessible via SQL endpoint
df.withColumn("ssn", col("raw_ssn")) \
    .write.format("delta").saveAsTable("silver.player_pii")
# SSN queryable by anyone with SQL endpoint access

# ✅ CORRECT: Hashed SSN in Tables, raw encrypted in Files (restricted)
import hashlib
import os

salt = os.environ.get("FABRIC_POC_HASH_SALT", "")
if not salt:
    raise ValueError("FABRIC_POC_HASH_SALT environment variable required")

@udf(StringType())
def hash_ssn(ssn):
    return hashlib.sha256(f"{salt}{ssn}".encode()).hexdigest()

# Tables: only the hash (for matching, not reversible)
df.withColumn("ssn_hash", hash_ssn(col("raw_ssn"))) \
    .drop("raw_ssn") \
    .write.format("delta").saveAsTable("silver.player_profile")

# Files: encrypted original (for compliance retrieval only)
# Stored in restricted Files folder with limited access
df.select("player_id", "raw_ssn") \
    .write.format("parquet") \
    .mode("overwrite") \
    .save("Files/restricted/player_pii/")

# Automated cleanup for landing zone
from datetime import datetime, timedelta

cutoff = datetime.now() - timedelta(days=30)
landing_files = mssparkutils.fs.ls("Files/landing/")

for item in landing_files:
    if item.isDir:
        sub_files = mssparkutils.fs.ls(item.path)
        for f in sub_files:
            mod_time = datetime.fromtimestamp(f.modifyTime / 1000)
            if mod_time < cutoff:
                mssparkutils.fs.rm(f.path, recurse=True)
                print(f"Cleaned: {f.path} (modified {mod_time})")

# Reading data from a shortcut in Files section
df_external = spark.read.format("parquet") \
    .load("Files/shortcuts/partner_data/transactions/")

# Promoting shortcut data to a managed Delta table
df_external.write.format("delta") \
    .mode("overwrite") \
    .saveAsTable("bronze.partner_transactions")
# Now available via SQL endpoint and Direct Lake