PySpark列值长度验证：生成错误标记列的实现方法求助

可以通过PySpark内置函数实现列长度验证，并将不符合规则的列名拼接成error列，具体实现步骤如下：

1. 导入依赖并创建示例DataFrame #

先导入PySpark函数库，再创建你提供的示例DataFrame：

from pyspark.sql import SparkSession
from pyspark.sql import functions as F

# 初始化SparkSession
spark = SparkSession.builder.appName("ColumnValidation").getOrCreate()

# 构造示例数据
data = [
    ("abcd", "gs", "abmj", "a"),
    ("hgj", "vxb", "basn", "f"),
    ("hg", "hj", "ndks", "dd")
]
df = spark.createDataFrame(data, ["a", "b", "c", "d"])

2. 定义验证规则生成错误标记 #

针对每列的长度规则，用when函数判断是否符合要求，不符合则返回列名，否则返回空字符串：

# 为各列生成验证标记列
df_with_flags = df \
    .withColumn("a_flag", F.when(F.length(F.col("a")) != 3, "a").otherwise("")) \
    .withColumn("b_flag", F.when(F.length(F.col("b")) > 2, "b").otherwise("")) \
    .withColumn("c_flag", F.when(F.length(F.col("c")) > 4, "c").otherwise("")) \
    .withColumn("d_flag", F.when(F.length(F.col("d")) != 1, "d").otherwise(""))

3. 拼接错误列并清理临时列 #

用concat_ws函数将非空的标记列拼接成error列，最后删除临时标记列：

# 拼接错误列名，用逗号分隔
result_df = df_with_flags \
    .withColumn("error", F.concat_ws(",", F.col("a_flag"), F.col("b_flag"), F.col("c_flag"), F.col("d_flag"))) \
    .drop("a_flag", "b_flag", "c_flag", "d_flag")

# 查看结果
result_df.show()

运行代码后得到的结果与你期望的一致：

+----+---+----+---+-----+
|   a|  b|   c|  d|error|
+----+---+----+---+-----+
|abcd| gs|abmj|  a|    a|
| hgj|vxb|basn|  f|    b|
|  hg| hj|ndks| dd|  a,d|
+----+---+----+---+-----+

补充说明 #

• 如果需要新增列的验证规则，只需添加对应的withColumn生成标记列，再将标记列加入concat_ws参数即可。
• concat_ws会自动忽略空字符串，因此只有不符合规则的列名会被拼接。

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