当前位置：首页 > news >正文

PySpark查询Dataframe中包含乱码的数据记录的方法

news 来源：原创 2025/5/8 7:45:40

首先，用PySpark获取Dataframe中所有非ASCII字符，找到其中的非乱码字符。

from pyspark.sql import SparkSession
from pyspark.sql.functions import col, concat_ws, explode, split, coalesce, lit
from pyspark.sql.types import StringType

spark = SparkSession.builder.appName("InvalidCharacterFinder").getOrCreate()

# 假设已存在DataFrame df
# df = ...

# 获取所有字符串类型列名
string_columns = [f.name for f in df.schema.fields if isinstance(f.dataType, StringType)]
result = []

if string_columns:
    # 处理空值并合并字符串列
    non_null_cols = [coalesce(col(c), lit("")).alias(c) for c in string_columns]
    combined_df = df.select(non_null_cols).select(concat_ws("", *string_columns).alias("merged_str"))
    
    # 拆分字符并过滤空字符串
    chars_df = combined_df.withColumn("char", explode(split(col("merged_str"), "")))\
                          .filter(col("char") != "")
    
    # 定义合法字符集合
    allowed_chars = set('abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789!'
                        '"#$%&\'()*+,-./:;<=>?@[]^_`{|}~')
    
    # 收集非法字符并去重
    unique_invalid = chars_df.rdd.map(lambda x: x.char)\
                             .filter(lambda c: c not in allowed_chars)\
                             .distinct()\
                             .collect()
    
    # 按首次出现顺序保留字符（分布式环境无法保证绝对顺序）
    seen = set()
    ordered_result = []
    for char in unique_invalid:
        if char not in seen:
            ordered_result.append(char)
            seen.add(char)
    result = ordered_result

print("非法字符集合:", ''.join(result))

代码说明：

数据准备：通过DataFrame Schema识别所有字符串类型的列
空值处理：使用coalesce函数将NULL转换为空字符串，确保后续字符串合并有效
列合并：使用concat_ws将多个字符串列的值合并为单个字符串
字符拆分：通过split+explode将字符串拆分为单个字符，并过滤空字符
非法字符过滤：使用RDD操作过滤不在白名单中的字符，并通过distinct去重
结果处理：使用有序集合保持字符首次出现的顺序（注意：分布式环境下无法保证绝对顺序）

注意事项：

最终结果字符顺序可能与实际数据中的出现顺序不完全一致
白名单包含94个可打印ASCII字符（排除空格和控制字符）
使用RDD操作提升分布式处理性能
最终结果字符串可能包含各类特殊符号、中文、表情符号等非标准ASCII字符

然后，将非乱码字符加入排除的字符中，用PySpark检查Dataframe中包含乱码字符的记录并导出Excel文件。

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 False
    return any(c not in allowed_chars for c in s)

has_invalid_udf = udf(has_invalid_chars, BooleanType())

# 构建过滤条件
if not string_columns:
    # 如果没有字符串列，直接创建空DataFrame
    result_df = spark.createDataFrame([], df.schema)
else:
    # 组合所有字符串列的检查条件
    condition = None
    for col_name in string_columns:
        col_condition = has_invalid_udf(col(col_name))
        if condition is None:
            condition = col_condition
        else:
            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()