Pyspark使用f.coalesce后如何保留列的原始数据类型

• 原始测试数据中使用字符串"NULL"表示空值，Spark无法识别为原生空值，会将string_value、numeric_value两列统一推断为字符串类型，连带其他列也可能被隐式转换为字符串。
• coalesce函数要求输入的所有列类型兼容，若同时输入字符串、数值两种不兼容类型，Spark会自动将所有输入隐式提升为字符串类型，最终输出结果全为字符串。

步骤1：修正原始数据的空值写法 #

将字符串"NULL"替换为Python原生None，让Spark正确识别空值，保留各列的原始数据类型，同时注意不要用list作为变量名，避免覆盖Python内置的list关键字：

from pyspark.sql import functions as f

id_list = ["B", "A", "D", "C"]

data = [("B", "On", None, 1632733508, "active"),
        ("B", "Off", None, 1632733508, "active"), 
        ("A","On", None, 1632733511, "active"),    
        ("A","Off", None, 1632733512, "active"),
        ("D", None, 450, 1632733513, "inactive"),
        ("D", None, 431, 1632733515, "inactive"),
        ("C", None, 20, 1632733518, "inactive"),
        ("C", None, 30, 1632733521, "inactive")]

df = spark.createDataFrame(data, ["unique_string", "ID", "string_value", "numeric_value", "timestamp","mode"])

步骤2：修改拆分函数避免类型隐式转换 #

同一个ID对应的行只会有string_value或numeric_value其中一列非空，所以可以先判断当前ID对应的取值列，直接选择对应列即可，不需要用coalesce拼接不同类型的列：

def split_df(df, list_val):
    filtered_df = df.filter(f.col('ID') == list_val)
    # 判断当前ID取值来自字符串列还是数值列
    has_string_val = filtered_df.filter(f.col('string_value').isNotNull()).count() > 0
    val_col = f.col('string_value') if has_string_val else f.col('numeric_value')
    return filtered_df.select(
        val_col.alias(list_val),
        f.col('timestamp'),
        f.col('mode')
    )

dfs = [split_df(df, id) for id in id_list]

效果验证 #

修改后执行dfs[2].printSchema()可看到类型正常保留：

root
 |-- D: long (nullable = true)
 |-- timestamp: long (nullable = true)
 |-- mode: string (nullable = true)

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