如何在PySpark/Python DataFrame中基于列检查重复数据并终止流程？

Hey there! Let’s tackle your two DataFrame duplicate-checking questions step by step—these are super common tasks, so I’ve got straightforward, efficient solutions for you.

PySpark gives you flexible ways to detect duplicates, whether you want a quick yes/no check or need to pinpoint exactly which rows are duplicated.

• Quick existence check (all columns)
  Compare the row count of your original DataFrame with the count after removing duplicates. If they don’t match, duplicates exist:

  original_row_count = df.count()
  distinct_row_count = df.dropDuplicates().count()
  
  if original_row_count != distinct_row_count:
      print("Duplicates detected in the DataFrame!")
  else:
      print("No duplicates found across all columns.")
  

• Check duplicates on specific columns
  If you only care about duplicates in a subset of columns (like a user ID and timestamp pair), pass those column names to dropDuplicates():

  target_columns = ["user_id", "transaction_date"]
  distinct_count = df.dropDuplicates(target_columns).count()
  
  if df.count() != distinct_count:
      print(f"Duplicates found in columns: {target_columns}")
  else:
      print(f"No duplicates in columns {target_columns}.")
  

• Find exactly which rows are duplicated
  To see the actual duplicate groups, use groupBy on your target columns and filter for groups with more than one entry:

  duplicate_groups = df.groupBy(target_columns).count().filter("count > 1")
  duplicate_groups.show()  # Displays all duplicate column combinations and their counts
  

For Pandas DataFrames, you can quickly check for duplicates on specified columns and terminate your workflow immediately if any are found—no need to waste resources processing further.

The most efficient method uses Pandas’ vectorized duplicated() function, optimized to run fast even on large datasets. Here’s how to implement it:

import pandas as pd
import sys

# Example DataFrame
df = pd.DataFrame({
    "customer_id": [101, 102, 103, 102],
    "order_date": ["2024-01-05", "2024-01-06", "2024-01-07", "2024-01-06"],
    "total": [250, 180, 320, 180]
})

# Define the columns to check for duplicates
check_columns = ["customer_id", "order_date"]

# Check if any duplicates exist in the target columns
if df.duplicated(subset=check_columns).any():
    print(f"Critical error: Duplicates detected in columns {check_columns}. Stopping process.")
    sys.exit(1)  # Terminates the script with a non-zero exit code (signals failure)

# If no duplicates, proceed with your workflow
print("No duplicates found. Continuing with data processing...")

Why this works efficiently: #

• duplicated() is vectorized (operates on entire columns at once) instead of looping through rows, making it way faster for large datasets.
• The .any() method stops checking as soon as it finds the first duplicate, so you don’t waste time scanning the entire DataFrame unnecessarily.

内容的提问来源于stack exchange，提问作者Prasanna Saraswathi Krishnan