使用PySpark DataFrame校验多邮箱列并按竖线分隔合并为单列

需求对齐 #

实现核心逻辑如下：

• 支持动态指定CSV读取后的多个邮箱字段，无需硬编码处理逻辑
• 先对每个字段内的所有邮箱（支持单个字段内用|分隔的多邮箱）单独做格式校验，过滤非法值
• 所有合法邮箱最终用|拼接为单列输出

调整后完整代码 #

from pyspark.sql.session import SparkSession
from pyspark.sql.functions import col, trim, split, array, flatten, filter, array_join, when

spark = SparkSession.builder.getOrCreate()

# 可配置参数，修改即可适配不同场景
input_csv_path = "file_name.csv"
# 要处理的邮箱列名，新增列直接往列表添加即可
email_cols = ["primary_email_addr", "alt_email_addr"]
# 邮箱校验正则，可根据实际业务规则调整
email_regex = r'^\w+([\.-]?\w+)*@\w+([\.-]?\w+)*(\.\w{2,3})+$'

# 读取CSV文件
df = spark.read.option("header", True).csv(input_csv_path)

# 核心处理逻辑
df_result = df.withColumn("all_emails", flatten(array([split(trim(col(c)), "\\|") for c in email_cols]))) \
              .withColumn("valid_emails", filter("all_emails", lambda x: x.rlike(email_regex))) \
              .withColumn("email_address", array_join("valid_emails", "|")) \
              .withColumn("email_addr_status", when(col("email_address") != "", "Valid").otherwise("Invalid")) \
              # 可选：删除中间临时列，不需要保留原始字段也可在这里指定drop
              .drop("all_emails", "valid_emails")

df_result.show(truncate=False)

核心调整说明 #

• 动态适配：将需要处理的邮箱字段抽成可配置列表，新增/修改处理字段只需调整email_cols参数，无需改动后续处理逻辑
• 逻辑顺序优化：先拆分单个字段内的多邮箱（适配示例中单个字段存储多个邮箱的场景），再逐个校验合法性，避免原代码合并后校验导致的非法值遗漏问题
• 性能优化：使用PySpark原生高阶数组函数处理，无需额外shuffle操作，处理效率更高
• 边界兼容：自动处理字段首尾空格、空值、非法邮箱的过滤，最终拼接结果只会保留合法邮箱

内容的提问来源于stack exchange，提问作者Naveen