看起来你遇到了两个核心困扰：一是CSV读取时欧元符号€变成乱码�，直接替换乱码根本不管用；二是数值列混着%、€和千分位逗号，没法直接做除法计算，还得处理除以0的异常情况。我来给你一套靠谱的解决方案：

问题根源拆解 #

1. 乱码问题：�是编码不匹配时的占位符，你用ISO-8859-1读取可能不对——欧元符号在UTF-8或Windows-1252（西欧常用编码）里才能正确解析，直接替换�是治标不治本，得从读取编码入手。
2. 数值计算问题：你的数值列里有非数字字符（%、€、逗号、空格），必须先清理干净转成数值类型，还要处理除数为0的情况，不然会触发运行时错误。

分步解决方案 #

1. 先搞定编码，正确读取CSV #

先换用正确的编码读取文件，优先试UTF-8；如果是Excel导出的CSV，大概率用Windows-1252：

# 优先尝试UTF-8编码
df_src = spark.read.csv(src_path, header=True, encoding='UTF-8')
# 如果还是乱码，换成Windows-1252编码
# df_src = spark.read.csv(src_path, header=True, encoding='Windows-1252')

这样读取后，€符号就能正常显示，不用再处理乱码�了。

2. 写个通用函数清理数值列 #

我们需要把%、€、空格去掉，千分位逗号也删掉，然后转成可计算的数值类型：

from pyspark.sql import functions as F
from pyspark.sql.types import DoubleType

def clean_numeric_column(col_name):
    # 移除€、%、空格字符
    cleaned = F.translate(F.col(col_name), "€% ", "")
    # 去掉千分位的逗号
    cleaned = F.regexp_replace(cleaned, ",", "")
    # 转成Double类型，转换失败自动返回null
    return F.cast(cleaned, DoubleType()).alias(f"{col_name}Replaced")

3. 计算ResultValue，处理除以0的情况 #

用when函数判断除数是否为0，避免报错，直接返回null：

df = df_src.select(
    'Month',
    'TrueValue',
    clean_numeric_column('TrueValue'),
    'PickoutValue',
    clean_numeric_column('PickoutValue')
).withColumn(
    'ResultValue',
    F.when(
        F.col('PickoutValueReplaced') == 0,  # 除数为0时返回null
        None
    ).otherwise(
        (F.col('TrueValueReplaced') / F.col('PickoutValueReplaced')) * 100
    )
)

# 查看最终结果
df.show(truncate=False)

完整可运行代码 #

from pyspark.sql import SparkSession
from pyspark.sql import functions as F
from pyspark.sql.types import DoubleType

# 初始化SparkSession
spark = SparkSession.builder.appName("CSV_Euro_Calculation").getOrCreate()

# 替换成你的CSV文件路径
src_path = "your_csv_file_path.csv"

# 1. 正确读取CSV，选择合适编码
df_src = spark.read.csv(src_path, header=True, encoding='UTF-8')
# 若UTF-8无效，替换为encoding='Windows-1252'

# 2. 定义数值清理函数
def clean_numeric_column(col_name):
    cleaned = F.translate(F.col(col_name), "€% ", "")
    cleaned = F.regexp_replace(cleaned, ",", "")
    return F.cast(cleaned, DoubleType()).alias(f"{col_name}Replaced")

# 3. 处理列并计算结果
df = df_src.select(
    'Month',
    'TrueValue',
    clean_numeric_column('TrueValue'),
    'PickoutValue',
    clean_numeric_column('PickoutValue')
).withColumn(
    'ResultValue',
    F.when(
        F.col('PickoutValueReplaced') == 0,
        None
    ).otherwise(
        (F.col('TrueValueReplaced') / F.col('PickoutValueReplaced')) * 100
    )
)

# 展示结果
df.show(truncate=False)

关键说明 #

• 编码选择：UTF-8是通用编码，但如果你的CSV是从Windows系统的Excel导出的，Windows-1252更适配西欧字符（包括欧元符号）。
• 数值转换：用cast(DoubleType())时，任何无法转换为数值的内容都会自动返回null，完全符合你预期的结果。
• 除以0处理：通过when分支判断，避免了运行时除以0的错误，直接返回null，和你想要的输出一致。

这样处理后，就能完美得到你预期的DataFrame啦！

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