PySpark DataFrame嵌套字典中Bytes转Decimal问题求助

你的问题核心是直接对整个JSON格式的字符串执行cast("binary")，这会把整个JSON文本转成二进制，而非提取嵌套的bytes字段内容再转换。以下是正确处理流程：

步骤1：解析JSON字符串，提取bytes字段 #

原字段是JSON结构的字符串，先通过from_json解析成结构体，取出目标bytes值：

from pyspark.sql import functions as F
from pyspark.sql.types import StructType, StructField, StringType

# 定义JSON结构
json_schema = StructType([
    StructField("bytes", StringType(), nullable=True)
])

# 解析JSON并提取bytes字段
df = df.withColumn("parsed_json", F.from_json(F.col("Col_name"), json_schema))
df = df.withColumn("bytes_str", F.col("parsed_json.bytes"))

步骤2：将转义Unicode字符串转为Binary类型 #

示例中的"\u0014ok\u0000"是用Unicode转义表示的字节序列，需要把这个字符串还原成真实二进制数据，通过自定义UDF实现：

def unicode_esc_to_binary(s):
    if not s:
        return None
    # 先解码Unicode转义，再转成字节
    return s.encode("utf-8").decode("unicode-escape").encode("latin-1")

# 注册UDF
esc_to_bin_udf = F.udf(unicode_esc_to_binary, "binary")

# 转换为binary类型
df = df.withColumn("binary_data", esc_to_bin_udf(F.col("bytes_str")))

步骤3：将Binary转为Decimal类型 #

根据实际的Decimal逻辑定义（精度、小数位数、端序），使用Spark的binary_to_decimal函数转换（Spark 3.0+支持）：

# 假设Decimal精度为10、小数位数为0，根据实际情况调整
df = df.withColumn("decimal_value", F.binary_to_decimal(F.col("binary_data"), 10, 0))

完整示例代码 #

from pyspark.sql import SparkSession
from pyspark.sql import functions as F
from pyspark.sql.types import StructType, StructField, StringType

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

# 模拟测试数据
data = [("""{"bytes":"\\u0014ok\\u0000"}""",)]
df = spark.createDataFrame(data, ["Col_name"])

# 解析JSON
json_schema = StructType([StructField("bytes", StringType())])
df = df.withColumn("parsed_json", F.from_json("Col_name", json_schema))
df = df.withColumn("bytes_str", F.col("parsed_json.bytes"))

# 转义字符串转Binary
def unicode_esc_to_binary(s):
    if not s:
        return None
    return s.encode("utf-8").decode("unicode-escape").encode("latin-1")

esc_to_bin_udf = F.udf(unicode_esc_to_binary, "binary")
df = df.withColumn("binary_data", esc_to_bin_udf("bytes_str"))

# Binary转Decimal
df = df.withColumn("decimal_value", F.binary_to_decimal("binary_data", 10, 0))

# 查看结果
df.select("Col_name", "binary_data", "decimal_value").show(truncate=False)

注意事项 #

• 如果字节序列是小端序，需在binary_to_decimal中指定endianness='little'参数（Spark 3.2+支持）。
• 务必根据实际业务调整Decimal的精度和小数位数。
• 若bytes字段是Base64编码，直接用F.unbase64(F.col("bytes_str")).cast("binary")替代UDF即可。

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