Databricks中解析含字符串型数值的JSON数组并转类型失败求助

你遇到的问题核心是JSON里的_value是带引号的字符串类型，Spark的from_json函数默认不会自动将字符串格式的数值转换为浮点类型，直接指定_value:float会导致解析失败返回null。以下是几种可行的解决方法：

方法1：先解析为字符串，再批量转换为浮点数 #

这是最稳妥的方案，先按字符串类型完成JSON解析，再用transform函数遍历数组结构体，将_value转换为精度更高的double类型：

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

bronze_data = spark.read.table(f"{catalog}.{schema}_bronze.{table}")

# 1. 先解析为包含字符串类型_value的数组结构体
parsed_df = bronze_data.select(
    F.from_json(
        F.col("_cost_per_action_type"),
        "array<struct<_action_type:string,_value:string>>"
    ).alias("cost_per_action_type")
)

# 2. 遍历数组，将每个元素的_value转为double类型
silver_data = parsed_df.select(
    F.transform(
        F.col("cost_per_action_type"),
        lambda x: F.struct(
            x["_action_type"].alias("_action_type"),
            x["_value"].cast(DoubleType()).alias("_value")
        )
    ).alias("cost_per_action_type")
)

silver_data.show(5, truncate=100)

方法2：使用自定义Schema+解析选项（Spark 3.0+） #

如果你的Spark版本在3.0及以上，可以通过指定解析选项allowStringToNumericConversion，让from_json自动将字符串数值转换为浮点类型：

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

# 定义目标Schema
target_schema = ArrayType(
    StructType([
        StructField("_action_type", StringType()),
        StructField("_value", DoubleType())
    ])
)

bronze_data = spark.read.table(f"{catalog}.{schema}_bronze.{table}")

silver_data = bronze_data.select(
    F.from_json(
        F.col("_cost_per_action_type"),
        target_schema,
        options={"allowStringToNumericConversion": "true"}
    ).alias("cost_per_action_type")
)

silver_data.show(5, truncate=100)

方法3：修复UDF实现（若需使用UDF） #

之前UDF失败大概率是写法问题，以下是正确的UDF实现：

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

# 定义单个结构体的转换逻辑
def convert_value_to_float(action_struct):
    if not action_struct:
        return None
    return {
        "_action_type": action_struct["_action_type"],
        "_value": float(action_struct["_value"]) if action_struct["_value"] else None
    }

# 定义UDF的返回类型
udf_return_type = StructType([
    StructField("_action_type", StringType()),
    StructField("_value", FloatType())
])

# 注册UDF
convert_udf = F.udf(convert_value_to_float, ArrayType(udf_return_type))

bronze_data = spark.read.table(f"{catalog}.{schema}_bronze.{table}")

# 先解析为字符串类型数组，再用UDF转换
silver_data = bronze_data.select(
    convert_udf(
        F.from_json(
            F.col("_cost_per_action_type"),
            "array<struct<_action_type:string,_value:string>>"
        )
    ).alias("cost_per_action_type")
)

silver_data.show(5, truncate=100)

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