如何用PySpark将Spark DataFrame中的0值替换为Null？

需求：将Spark DataFrame中所有列的0值替换为Null，导出JSON时避免输出0值属性；由于使用pivot动态生成列，无法逐列指定处理。

原始数据 #

+----------------------+---------------+----------+------------+-------------+--------------------+--------------+     
|ID                |T1             |T2        |T3          |T4           |T5                 |DT             |         
+----------------------+---------------+----------+------------+-------------+--------------------+--------------+     
|929916248484355237|0.000000       |0.0       |0E-10       |20.0000000000|20.0                |20220916120738|         
|772216248474350399|0.880000       |0.0       |1.4808200000|0E-10        |2.36082             |20220916120738|         
|772216248474350399|0.000000       |0.0       |0E-10       |20.0000000000|20.0                |20220916120738|         
|075616248464351729|0.010000       |0.0       |0.0127000000|0E-10        |0.022699999999999998|20220916120738|         
|915716248424355578|0.010000       |0.0       |0.0127000000|0E-10        |0.022699999999999998|20220916120738|         
|277016248484357606|0.000000       |0.0       |0E-10       |20.0000000000|20.0                |20220916120738|         
|739516248574350647|0.000000       |0.0       |0E-10       |20.0000000000|20.0                |20220916120738|         
|915716248424355578|0.000000       |0.0       |0E-10       |20.0000000000|20.0                |20220916120738|         
|075616248464351729|0.000000       |0.0       |0E-10       |20.0000000000|20.0                |20220916120738|         
|756216248594352650|0.000000       |0.0       |0E-10       |20.0000000000|20.0                |20220916120738|         
+----------------------+---------------+----------+------------+-------------+--------------------+--------------+     

期望结果 #

+----------------------+---------------+----------+------------+-------------+--------------------+--------------+     
|ID                |T1             |T2        |T3          |T4           |T5                  |DT             |         
+----------------------+---------------+----------+------------+-------------+--------------------+--------------+     
|929916248484355237|null           |0.0       |null        |20.0000000000|20.0                |20220916120738|         
|772216248474350399|0.880000       |0.0       |1.4808200000|null         |2.36082             |20220916120738|         
|772216248474350399|null           |0.0       |null        |20.0000000000|20.0                |20220916120738|         
|075616248464351729|0.010000       |0.0       |0.0127000000|null         |0.022699999999999998|20220916120738|         
|915716248424355578|0.010000       |0.0       |0.0127000000|null         |0.022699999999999998|20220916120738|         
|277016248484357606|null           |0.0       |null        |20.0000000000|20.0                |20220916120738|         
|739516248574350647|null           |0.0       |null        |20.0000000000|20.0                |20220916120738|         
|915716248424355578|null           |0.0       |null        |20.0000000000|20.0                |20220916120738|         
|075616248464351729|null           |0.0       |null        |20.0000000000|20.0                |20220916120738|         
|756216248594352650|null           |0.0       |null        |20.0000000000|20.0                |20220916120738|         
+----------------------+---------------+----------+------------+-------------+--------------------+--------------+    

已实现的可行代码 #

cols = [when(~col(x).isin(0), col(x)).alias(x) for x in df.columns]
df = df.select(*cols)

更优实现方案 #

1. 使用selectExpr简化表达式写法 #

直接通过SQL风格的字符串表达式生成处理逻辑，代码更简洁：

exprs = [f"CASE WHEN {col} != 0 THEN {col} ELSE NULL END AS {col}" for col in df.columns]
df = df.selectExpr(*exprs)

优点：无需引入when/col函数，写法贴近原生SQL，可读性强。

2. 仅处理数值类型列（跳过非数值列） #

如果DataFrame包含非数值类型列（如示例中的ID、DT），可以只对数值列做处理，减少无效计算：

from pyspark.sql.types import DoubleType, FloatType, IntegerType

numeric_cols = [col.name for col in df.schema if isinstance(col.dataType, (DoubleType, FloatType, IntegerType))]
cols = [when(~col(x).isin(0), col(x)).alias(x) if x in numeric_cols else col(x) for x in df.columns]
df = df.select(*cols)

优点：避免对非数值列执行判断逻辑，提升处理效率。

3. 使用foldLeft链式处理列 #

通过链式调用逐步修改每个列，无需提前构建列列表，适合大量动态列场景：

from pyspark.sql.functions import when, col

df_processed = df.columns.foldLeft(df, lambda acc, c: acc.withColumn(c, when(~col(c).isin(0), col(c))))

优点：代码紧凑，直接对DataFrame进行链式修改，逻辑连贯。

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