from pyspark.sql import SparkSession
from pyspark.sql.types import StringType
from pyspark.sql.functions import udf, col
from pyspark.sql.types import BooleanType
import pandas as pd# 初始化Spark会话
spark = SparkSession.builder.appName("StringFilter").getOrCreate()# 假设原始DataFrame为df（这里需要替换为实际的数据源读取逻辑）
# df = spark.read.csv("input.csv", header=True)# 获取所有字符串类型的列名
string_columns = [field.name for field in df.schema.fields if isinstance(field.dataType, StringType)]# 定义允许的字符集合
allowed_chars = set('abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789!"#$%&\'()*+,-./:;<=>?@[]^_`{|}~')# 定义检查非法字符的UDF
def has_invalid_chars(s):if s is None:return Falsereturn any(c not in allowed_chars for c in s)has_invalid_udf = udf(has_invalid_chars, BooleanType())# 构建过滤条件
if not string_columns:# 如果没有字符串列，直接创建空DataFrameresult_df = spark.createDataFrame([], df.schema)
else:# 组合所有字符串列的检查条件condition = Nonefor col_name in string_columns:col_condition = has_invalid_udf(col(col_name))if condition is None:condition = col_conditionelse:condition = condition | col_condition# 过滤出包含非法字符的行filtered_df = df.filter(condition)# 创建结构相同的空DataFrame并合并结果empty_df = spark.createDataFrame([], df.schema)result_df = empty_df.union(filtered_df)# 导出为Excel文件（注意：此操作会将数据收集到Driver节点）
pd_df = result_df.toPandas()
pd_df.to_excel("output.xlsx", index=False)# 停止Spark会话（根据实际需要决定是否保留会话）
spark.stop()

代码说明：

1. 初始化与数据读取：需要根据实际数据源替换读取逻辑（示例中被注释掉的spark.read.csv部分）
2. 获取字符串列：通过分析Schema获取所有字符串类型的字段
3. 定义字符白名单：使用集合类型提升查询效率
4. UDF定义：用于检查字符串是否包含非法字符
5. 条件构建：使用逻辑或组合所有字符串列的检查条件
6. 结果处理：
   • 直接处理空字符串列的边界情况
   • 使用union保持与原DataFrame结构一致
7. Excel导出：
   • 通过转换为Pandas DataFrame实现
   • 注意大数据量时可能存在的内存问题

注意事项：

1. 大数据量场景下建议分批次处理或使用分布式写入方式
2. Excel导出操作会触发数据收集到Driver节点，需确保资源充足
3. 实际应用中建议添加异常处理机制
4. 空值处理逻辑可根据业务需求调整（当前版本忽略NULL值）