How can you validate email addresses in a DataFrame using PySpark?

Normalise the value (trim spaces, lower-case), then test it against a practical regular expression with rlike, and flag or quarantine rows that fail. A regular expression checks the format only: it can't tell you whether the mailbox exists, so treat it as a first filter.

from pyspark.sql import functions as F

EMAIL_RE = r"^[A-Za-z0-9._%+-]+@[A-Za-z0-9-]+(\.[A-Za-z0-9-]+)*\.[A-Za-z]{2,}$"

checked = (df
    .withColumn("email_clean", F.lower(F.trim("email")))
    .withColumn("email_valid", F.col("email_clean").rlike(EMAIL_RE)))

valid = checked.filter("email_valid")
invalid = checked.filter(~F.col("email_valid") | F.col("email_valid").isNull())

On sample data, " Ann@Example.com " becomes ann@example.com and passes; bo@x (no domain ending) and cy@@x.com fail; di@mail.co.uk passes; a null email gives a null result, which is why the invalid filter checks for null too.

About the pattern

• It requires one @, a local part of letters, digits and ._%+-, a domain of labels separated by dots, and a final part of at least 2 letters.
• The full standard for email addresses (RFC 5322) allows things like quoted local parts that almost no real system uses; a stricter regular expression that follows the standard exactly is long and rarely helpful.
• Internationalised addresses (non-ASCII letters) fail this pattern; widen it if your users have them.
• Lower-casing is safe for the domain; the local part is technically case-sensitive, but practically every provider treats it as case-insensitive.

Going further

• Extra checks with column expressions: length under 254 characters, no consecutive dots (~F.col("email_clean").contains("..")), domain in a list of known disposable-email domains.
• Library-based validation in a pandas UDF (for example email-validator), which can also check the domain's DNS records, but that is slow and needs network access from executors; do it outside the main pipeline if at all.
• Real verification only comes from sending a confirmation email; that belongs in the application, not the data pipeline.

Handling invalid values Don't silently drop them: keep the original value, a validity flag and a reason, write failures to a quarantine table, and count them over time. On Databricks, an expectation in a declarative pipeline (expect_or_drop with an RLIKE condition) or a Delta CHECK constraint enforce the rule in the table itself.