在 Pyspark 中将多行作为 JSON 对象与定义的模式组合

Question

我想对数据框执行以下操作。对于具有相似 id(L_ID) 的行，我需要通过传递定义的模式将关联的列组合成一个 JSON。如下面的示例所示，输出应该是 1 条 JSON 消息，其他列具有定义的模式。

我得到了这个样品 df:

我已经定义了一个模式：

按“年龄”列对行进行分组

grouped_df = df.groupBy('L_ID').agg(collect_list(struct('L_ID','L_NAME', 'L_DESC', 'C_VERSION', 'S_ID1', 'S_NAME', 'TIMEZONE', 'A', 'C_ID', 'ID', 'VER', 'S_ID2')).alias('json_column'))

转换为 json 字符串

result = grouped_df.selectExpr("to_json(json_column) AS json")

提供的模式以定义的模式生成 json

result1 = result.withColumn('json_column_with_schema', from_json('json', schema))

但我看到的实际结果是显示（result1）时

我正在寻找低于预期的结果，