Spark Dataframe如何处理■、�、□等NUL类型的无效乱码字符

你遇到的这类乱码包含两类字符，一类是源端存储的ASCII 0x00空字符（NUL），一类是编码解析失败时产生的Unicode替换字符�，不需要调整编码配置，直接在DataFrame层通过字符处理函数即可清理，处理逻辑完全兼容流式作业的语义要求。

具体实现步骤 #

1. 先将Event Hub读取的二进制字段正确转码为字符串

Event Hub返回的body字段默认为BinaryType，转码时显式指定无效字符的处理逻辑，避免二次乱码

from pyspark.sql import SparkSession
from pyspark.sql.functions import col, decode, regexp_replace

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

# 读取Event Hub流数据，eh_conf为你的Event Hub连接配置
df = spark \
  .readStream \
  .format("eventhubs") \
  .options(**eh_conf) \
  .load()

# 二进制字段转字符串，解析失败的字节统一替换为�，方便后续统一清理
df = df.withColumn("raw_str", decode(col("body"), "UTF-8", "replace"))

2. 清理异常字符

• 方法1：针对性清理NUL字符和替换乱码，适合仅需处理这两类异常的场景

# 替换所有\x00（NUL）和�为空
clean_df = df.withColumn("clean_str", regexp_replace(col("raw_str"), r"[\x00�]", ""))

• 方法2：仅保留合法可打印字符，适合有大量不可见控制字符的场景

# 示例保留ASCII可打印字符、中文、常见全角符号，其余字符全部删除，可根据业务需求扩展正则范围
clean_df = df.withColumn("clean_str", regexp_replace(col("raw_str"), r"[^\x20-\x7E\u4e00-\u9fa5\uff01-\uff5e]", ""))

3. 效果验证及上线
   可以先通过控制台输出验证清理效果，确认无误后再替换为实际的输出逻辑

query = clean_df \
  .writeStream \
  .format("console") \
  .option("truncate", "false") \
  .start()

query.awaitTermination()

注意事项 #

• 若NUL仅出现在字符串首尾作为填充字符，可仅替换首尾空字符，性能优于全量替换：regexp_replace(col("raw_str"), r"^\x00+|\x00+$", "")
• 所有处理函数均为Spark原生的确定性函数，不会影响流作业的容错性和Exactly-Once语义

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