您可以使用下面的代码在单独的行中创建，并将数据写入message_records和messages的单独文件中。

record = {}
record["field1"] = json_data["field1"]
record["field2"] = json_data["field2"]
message_records_df =spark.createDataFrame([record])
messages_df = spark.createDataFrame([record])

使用field1和field2创建 Dataframe 。因为这两个都是message_records和messages。

from pyspark.sql.types import LongType
from pyspark.sql import Row

def zipindexdf(df):
    schema_new = df.schema.add("index", LongType(), False)
    return df.rdd.zipWithIndex().map(lambda l: list(l[0]) + [l[1]]).toDF(schema_new)
    
message_records_df_index = zipindexdf(message_records_df)
message_records_df_index.show()
messages_df_index = zipindexdf(messages_df)
messages_df_index.show()

这里我使用zipWithIndex添加index列，以便它可以在索引上连接。

接下来，创建dataframe并通过循环message_records中的每个项目合并到最终的dataframe，如下所示。

for i in json_data['message_records']:
    df = zipindexdf(spark.createDataFrame([i]))
    message_records_df_index = message_records_df_index.join(df, "index", "inner")
    message_records_df_index.show()

与下面的消息相同

for i in json_data['messages']:
    df = zipindexdf(spark.createDataFrame([i]))
    messages_df_index = messages_df_index.join(df, "index", "inner")
    messages_df_index.show()

最后，将这些数据写入csv文件。

message_records_df_index.write.option("header","true").csv('/message_records_df/')
messages_df_index.write.option("header","true").csv('/messages_df/')

spark.read.option("header","true").csv('/message_records_df/').show()
spark.read.option("header","true").csv('/messages_df/').show()

你可以在this SO answer中找到答案。
你只需要改变你调用paras的方式。我以前

df_flat1 = flatten_test(multiline_df.select(multiline_df.field1,multiline_df.field2,multiline_df.message_records))
df_flat2 = flatten_test(multiline_df.select(multiline_df.field1,multiline_df.field2, multiline_df.messages))
df_flat1.printSchema()
df_flat1.show(5)

df_flat2.printSchema()
df_flat2.show(5)

得到了

root
 |-- field1: string (nullable = true)
 |-- field2: string (nullable = true)
 |-- message_records_field3: string (nullable = true)
 |-- message_records_field4: string (nullable = true)
 |-- message_records_field5: string (nullable = true)
 |-- message_records_field6: string (nullable = true)

+------+------+----------------------+----------------------+----------------------+----------------------+
|field1|field2|message_records_field3|message_records_field4|message_records_field5|message_records_field6|
+------+------+----------------------+----------------------+----------------------+----------------------+
|value1|value2|                value3|                value4|                  null|                  null|
|value1|value2|                  null|                  null|                value5|                value6|
+------+------+----------------------+----------------------+----------------------+----------------------+

root
 |-- field1: string (nullable = true)
 |-- field2: string (nullable = true)
 |-- messages_field10: string (nullable = true)
 |-- messages_field11: string (nullable = true)
 |-- messages_field12: string (nullable = true)
 |-- messages_field7: string (nullable = true)
 |-- messages_field8: string (nullable = true)
 |-- messages_field9: string (nullable = true)

+------+------+----------------+----------------+----------------+---------------+---------------+---------------+
|field1|field2|messages_field10|messages_field11|messages_field12|messages_field7|messages_field8|messages_field9|
+------+------+----------------+----------------+----------------+---------------+---------------+---------------+
|value1|value2|            null|            null|            null|         value3|         value4|           null|
|value1|value2|          value6|            null|            null|           null|           null|         value5|
|value1|value2|            null|          value5|          value6|           null|           null|           null|
+------+------+----------------+----------------+----------------+---------------+---------------+---------------+