将 PySpark DataFrame 序列化为 JSON 数组答案

【问题标题】：Serialize PySpark DataFrame as JSON array将 PySpark DataFrame 序列化为 JSON 数组
【发布时间】：2019-12-17 15:26:27
【问题描述】：

我在使用 PySpark 的 Azure Databricks 中有一个 DataFrame。我需要将它作为 JSON 序列化为一个或多个文件。这些文件最终会上传到 Cosmos，因此格式正确的 JSON 至关重要。

我知道如何直接连接到 Cosmos 以直接序列化数据，但我需要创建 JSON 文件以便稍后上传到 Cosmos。

我无法从我的实际 DataFrame 中提供数据，但结构很复杂。每行都有嵌入的对象，其中一些有自己的嵌入对象和对象数组。

我认为问题在于我如何尝试序列化数据，而不是我如何转换它。我创建了这个简单的 DataFrame，df，我认为这足以作为示例。

+---------+-------------+
|property1|       array1|
+---------+-------------+
|   value1|["a","b","c"]|
|   value2|["x","y","z"]|
+---------+-------------+

我像这样将它序列化到 Azure Data Lake Storage Gen2。

df.coalesce(1).write.json(outpath, lineSep=",")

该文件将包含此 JSON。这些行不是数组中的元素，并且最后一行有一个尾随逗号，因此此 JSON 不会与 Cosmos 合作。

{"property1":"value1","array1":["a","b","c"]},
{"property1":"value2","array1":["x","y","z"]},

此 JSON 按预期上传。

[{"property1":"value1","array1":["a","b","c"]},
{"property1":"value2","array1":["x","y","z"]}]

我已成功上传单个 JSON 对象（即没有 [] 将它们括起来），因此任何将每个 DataFrame 行写入其自己的 JSON 文件的解决方案都是潜在的赢家。

我已经通过 repartitioning 尝试过，但总是有包含多行的文件。

【问题讨论】：

标签： json dataframe pyspark azure-cosmosdb azure-databricks

【解决方案1】：

我想出了两种方法。

首先使用 df.toJSON().collect() 创建 JSON 字符串行列表，将数组切片成批次，然后构建 JSON 数组字符串。

def batchWriteDataFrame(dataframe):
  rows = dataframe.toJSON().collect()
  batches = [rows[i * batch_size:(i + 1) * batch_size] for i in range((len(rows) + batch_size - 1) // batch_size)] # slice the rows into batches
  batch_num = 1

  for batch in batches:
    dbutils.fs.put(outpath + "batch/" + str(batch_num) + ".json", "[" + ",".join([row for row in batch]) + "]")
    batch_num += 1

第二个将每一行写入自己的文件。

def writeDataFrameRows(dataframe):
  i = 0
  for row in dataframe.toJSON().collect():
    dbutils.fs.put(outpath + "single/" + str(i) + ".json", row)
    i += 1

【讨论】：