PySpark动态传参至countDistinct及日期格式校验计数问题

一、处理Unique规则动态传参的AnalysisException错误 #

当动态传入多列执行countDistinct时出现AnalysisException，通常是列名拼接语法错误、列名不存在或动态表达式未被正确解析导致。以下是基于Spark的解决方案：

1. 读取规则并提取目标列
   先从规则CSV中筛选出Rule=Unique的行，收集对应的列名列表：

from pyspark.sql import SparkSession

spark = SparkSession.builder.appName("RuleCheck").getOrCreate()
rules_df = spark.read.csv("rules.csv", header=True, inferSchema=True)

# 收集Unique规则对应的列名
unique_col_list = [row.ColumnName for row in rules_df.filter(rules_df.Rule == "Unique").collect()]

2. 安全构建动态countDistinct表达式
   避免直接硬拼字符串引发语法错误，使用selectExpr或SQL语句解析动态表达式，同时处理列名含特殊字符的情况：

# 对列名添加反引号，防止特殊字符导致语法问题
quoted_unique_cols = [f"`{col}`" for col in unique_col_list]

if quoted_unique_cols:
    # 动态生成count(DISTINCT col1, col2...)表达式
    distinct_count = spark.sql(f"""
        SELECT count(DISTINCT {', '.join(quoted_unique_cols)}) as distinct_count
        FROM target_table
    """).collect()[0][0]
    # 或使用DataFrame API实现
    # distinct_count = df.selectExpr(f"count(DISTINCT {', '.join(quoted_unique_cols)})").collect()[0][0]
else:
    distinct_count = 0

3. 关键检查点

• 确认规则CSV中的ColumnName与目标数据表的列名完全一致（大小写敏感）
• 若列名包含空格、特殊字符，必须用反引号包裹
• 确保动态拼接的SQL表达式无多余逗号或语法错误

二、统计INSTALLDATE列不符合指定格式的记录数 #

假设RuleDetails中存储的是日期格式字符串（如yyyy-MM-dd），利用to_date函数解析失败返回null的特性，统计不符合格式的记录：

1. 获取日期格式规则
   从规则CSV中提取INSTALLDATE对应的格式要求：

# 筛选INSTALLDATE的格式规则
date_format = rules_df.filter(
    (rules_df.ColumnName == "INSTALLDATE") & (rules_df.Rule == "DateFormat")
).select("RuleDetails").collect()[0][0]

2. 统计不符合格式的记录
   过滤出to_date转换失败的行，并统计数量：

from pyspark.sql.functions import to_date

# 统计：排除INSTALLDATE自身为null的记录（若NotNull规则单独校验）
invalid_date_count = df.filter(
    df.INSTALLDATE.isNotNull() & to_date(df.INSTALLDATE, date_format).isNull()
).count()

# 若需包含null值的情况，直接过滤to_date结果为null的行：
# invalid_date_count = df.filter(to_date(df.INSTALLDATE, date_format).isNull()).count()

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