Write a transformation in PySpark to join and clean multiple raw input sources

Why It Matters (Architectural Logic): Multi-source joins require consistent keys, null handling, and skew mitigation. Netflix-scale = partition by business dims, broadcast small tables.

Join and clean multiple sources with consistent keys and null handling:

df1_clean = df1.dropDuplicates(["id"]).na.fill({"region": "Unknown"})
df2_clean = df2.filter(F.col("amount") > 0).dropDuplicates(["id"])
joined = df1_clean.join(df2_clean, "id", "left").drop(df2_clean.id)
final = joined.withColumn("valid", F.when(F.col("amount").isNull(), False).otherwise(True))

Best practices: use consistent join keys; handle skew with salt/broadcast; validate row counts before/after; log join key match rates; use coalesce for nullable columns; define cleaning rules as config. For Netflix-scale: partition by business dimensions, avoid cross-joins, use broadcast for small dimension tables.

Scalability Trade-offs: Broadcast small dims; salt for skew. Validate row counts; log join match rate. Avoid cross-joins.

Cost Implications: Shuffle on large tables = 70%+ of cost. Broadcast = 10-100x faster when applicable.