如何用Python和PySpark处理DataFrame中的特殊字符乱码问题

一、优先排查编码读取错误 #

绝大多数这类乱码是文件编码与读取编码不匹配导致的（比如原文件是UTF-8，却用Latin-1读取，特殊字符被替换为¿），先从根源解决：

Python 原生处理 #

尝试用正确编码重新加载文件，若已读取乱码数据，可通过编码转译修复：

import pandas as pd

# 尝试直接用UTF-8读取
try:
    df = pd.read_csv("your_file.csv", encoding="utf-8")
except UnicodeDecodeError:
    # 先以Latin-1读取（保留原始字节），再转译为UTF-8
    df = pd.read_csv("your_file.csv", encoding="latin-1")
    # 对所有字符串列批量转码
    str_cols = df.select_dtypes(include=["object"]).columns
    df[str_cols] = df[str_cols].apply(lambda col: col.str.encode("latin-1").str.decode("utf-8"))

验证修复效果：检查españa、algodón等词汇是否恢复正常。

PySpark 处理 #

读取时指定正确编码，或对已加载的乱码DataFrame进行转码：

from pyspark.sql import SparkSession
from pyspark.sql.functions import udf
from pyspark.sql.types import StringType

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

# 方式1：读取时指定正确编码
df = spark.read.csv("your_file.csv", header=True, encoding="utf-8")

# 方式2：对已读取的乱码列转码
def fix_encoding(s):
    return s.encode("latin-1").decode("utf-8") if s else s

fix_udf = udf(fix_encoding, StringType())
df_fixed = df.withColumn("target_column", fix_udf(df["target_column"]))

二、编码修复无效时，用上下文/字典修复 #

如果文件本身存储时已丢失字符（编码转译无法恢复），可基于业务场景的词汇映射或模糊匹配修复：

Python 原生处理 #

1. 预定义字典替换：针对高频损坏词汇构建映射表

fix_mapping = {
    "espa¿a": "españa",
    "algod¿on": "algodón",
    # 补充更多业务场景下的损坏-正确词汇对
}

# 批量替换指定列
df["target_column"] = df["target_column"].replace(fix_mapping, regex=False)

2. 模糊匹配修复：用fuzzywuzzy匹配正确词汇库（需先准备领域词汇表）

from fuzzywuzzy import process

# 正确词汇参考库
valid_words = ["españa", "algodón", "maíz", "niño"]

def fuzzy_fix(s):
    if "¿" in s:
        match, score = process.extractOne(s, valid_words)
        return match if score > 80 else s  # 设定匹配阈值
    return s

df["target_column"] = df["target_column"].apply(fuzzy_fix)

PySpark 处理 #

1. 广播字典批量替换：适合分布式场景下的高效替换

from pyspark.sql.functions import udf, broadcast

fix_mapping = {
    "espa¿a": "españa",
    "algod¿on": "algodón"
}

# 广播字典到所有节点，减少数据传输
broadcast_map = broadcast(spark.sparkContext.broadcast(fix_mapping))

def map_fix(s):
    return broadcast_map.value.get(s, s) if s else s

map_fix_udf = udf(map_fix, StringType())
df_fixed = df.withColumn("target_column", map_fix_udf(df["target_column"]))

2. 分布式模糊匹配：可结合Spark NLP的预训练模型或自定义UDF（需确保集群节点安装依赖）

三、复杂场景下的语言模型修复 #

如果上述方法都无法覆盖，可使用西班牙语预训练语言模型进行智能填充：

from transformers import pipeline

# 加载西班牙语BERT填充模型
fill_mask = pipeline("fill-mask", model="dccuchile/bert-base-spanish-wwm-uncased")

def llm_fix(s):
    if "¿" in s:
        masked_text = s.replace("¿", "[MASK]")
        top_result = fill_mask(masked_text)[0]
        return top_result["sequence"].replace("[CLS]", "").replace("[SEP]", "").strip()
    return s

df["target_column"] = df["target_column"].apply(llm_fix)

PySpark中可将此逻辑封装为UDF，注意集群环境需统一安装transformers库，或使用Spark NLP的原生模型更适配分布式计算。

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