Databricks视图中JSON列转换为API兼容格式的方法

方法1：直接提取原始JSON列（推荐） #

如果视图中存储JSON的列（假设列名为json_col）本身是合法JSON字符串，无需扁平化DataFrame，直接解析即可：

import json
from datetime import date
from pyspark.sql.functions import col

# 读取目标视图
df = spark.sql("SELECT json_col FROM your_view")

# 提取单条JSON字符串（批量处理可遍历collect结果）
raw_json = df.select(col("json_col")).first()[0]

# 解析为Python原生结构
data = json.loads(raw_json)

# 递归转换日期类型为API兼容的字符串格式
def format_date(obj):
    if isinstance(obj, dict):
        return {k: format_date(v) for k, v in obj.items()}
    elif isinstance(obj, list):
        return [format_date(i) for i in obj]
    elif isinstance(obj, date):
        return obj.isoformat()
    return obj

processed_data = format_date(data)

# 转成可直接用于API请求的JSON字符串
api_json = json.dumps(processed_data)

方法2：处理已扁平化的Row结果 #

如果已经完成扁平化操作得到包含Row对象的结果，可通过递归转换将Row转为Python原生字典：

import json
from datetime import date

def row2dict(row):
    result = {}
    for field in row.__fields__:
        val = getattr(row, field)
        if isinstance(val, date):
            result[field] = val.isoformat()
        elif hasattr(val, "__fields__"):  # 处理嵌套Row
            result[field] = row2dict(val)
        elif isinstance(val, list) and val and hasattr(val[0], "__fields__"):  # 处理Row列表
            result[field] = [row2dict(item) for item in val]
        else:
            result[field] = val
    return result

# 假设已通过collect()获取Row列表
rows = df.collect()
# 转换单条数据
api_data = row2dict(rows[0])
# 批量转换多条数据
# api_data_list = [row2dict(row) for row in rows]

# 生成API可用的JSON字符串
api_json = json.dumps(api_data)

批量处理优化方案 #

如果数据量较大，避免直接使用collect()，可通过Spark的toJSON()方法批量生成JSON字符串：

# 直接将DataFrame转为JSON格式的RDD
json_rdd = df.toJSON()
# 获取所有JSON字符串
json_list = json_rdd.collect()
# 每个字符串可直接用于API请求，或按需解析调整格式

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