The isin() function in PySpark is inefficient for large datasets. It builds a list of values in memory and performs filtering, which becomes resource-intensive and doesn't leverage distributed computation effectively. Replacing it with JOIN or MERGE operations ensures distributed processing and better performance.

Instead of using isin(), create a table with the records you want to delete and use a MERGE operation with the main Delta table. For smaller datasets, consider broadcasting the smaller table for efficiency. Here’s an example:

from delta.tables import DeltaTable

# Load records to delete into a temporary table
records_to_delete = spark.read.format("jdbc").option("query", "SELECT pk FROM sql_table").load()
records_to_delete.createOrReplaceTempView("delete_records")

# Use Delta's MERGE operation
delta_table = DeltaTable.forName(spark, "delta_table_name")
delta_table.alias("main").merge(
    records_to_delete.alias("delete"),
    "main.pk = delete.pk"
).whenNotMatchedDelete().execute()

For more details, refer to the Databricks documentation on Delta Lake MERGE and Optimizing Delta Tables. These resources provide additional guidance on using Delta Lake effectively for large-scale data management.