读取 json 加载到 pyspark 数据帧 [关闭]答案

【问题标题】：Read json load to pyspark dataframe [closed]读取 json 加载到 pyspark 数据帧 [关闭]
【发布时间】：2022-01-21 15:01:28
【问题描述】：

我想从 azure cosmos db 中提取数据，我正在使用 python sdk 在数据块中进行连接。

我希望能够将我的 json.load(data) 保存到 pyspark 数据帧中，因为我需要将数据保存在 databricks delta 湖中，如何将这些数据读取到 pyspark 数据帧。下面是我的代码和示例数据

{
 "appUuid": "aaaa-bbbb-cccc",
 "SystemId": null,
 "city": "Lancaster",
 "state": "NY",
 "zipCode": "140",
 "field1": "others",
 "field2": "others"
}
{
 "appUuid": "bbbb-dddd-eeee",
 "SystemId": null,
 "city": "Alden ",
 "state": "NY",
 "zipCode": "140",
 "field1": "others",
 "field2": "others"
}

from azure.cosmos import CosmosClient

client = CosmosClient('https://<cosmos_client>.documents.azure.com:443/', credential='AccountKey')
DATABASE_NAME = 'TestDB'
database = client.get_database_client(DATABASE_NAME)
CONTAINER_NAME = 'Test'
container = database.get_container_client(CONTAINER_NAME)

import json
for item in container.query_items(
         query='SELECT Top 10 * FROM Test',
        enable_cross_partition_query=True):
    data = json.dumps(item, indent=True)
    print(data)
    print(type(data))

# converting string to json dict
data1 = json.loads(data)
print(data1)
print(type(data1))


from pyspark.sql import *
from pyspark.sql import SparkSession,Row
spark = SparkSession.builder.getOrCreate()

df = spark.read.json(data1) -- I am getting error on this line.
display(df)

我收到此错误：

"IllegalArgumentException: java.net.URISyntaxException: Relative path in absolute URI: {"

【问题讨论】：

错误来自哪一行？
错误可以来自这一行 df = spark.read.json(data1) 我可以将数据读入 pyspark 数据帧
@gregeal - 请用这些细节编辑你的问题，而不是把它们放在 cmets 中。另外：究竟是什么被传递给spark.read.json() - 参数不是应该是文件名而不是数据吗？我建议在这方面多花点时间。
@DavidMakogon，请问如何将我的 json.dumps 保存为 databricks 文件系统（dbfs）中的 json 文件？

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

【解决方案1】：

您需要正确解析 JSON 数据

data = """{
 "appUuid": "aaaa-bbbb-cccc",
 "SystemId": null,
 "city": "Lancaster",
 "state": "NY",
 "zipCode": "140",
 "field1": "others",
 "field2": "others"
}
{
 "appUuid": "bbbb-dddd-eeee",
 "SystemId": null,
 "city": "Alden ",
 "state": "NY",
 "zipCode": "140",
 "field1": "others",
 "field2": "others"
}"""

import re  
#convert null for parsing
data = re.findall('[^}]+}',data.replace("null",'"null"'))

#cast to dict type
import ast
data = list(map(ast.literal_eval,data))

df = spark.read.json(sc.parallelize(data))
df.show()

输出：

+--------+--------------+---------+------+------+-----+-------+
|SystemId|       appUuid|     city|field1|field2|state|zipCode|
+--------+--------------+---------+------+------+-----+-------+
|    null|aaaa-bbbb-cccc|Lancaster|others|others|   NY|    140|
|    null|bbbb-dddd-eeee|   Alden |others|others|   NY|    140|
+--------+--------------+---------+------+------+-----+-------+

【讨论】：

感谢@Yefet，我能够修改代码以读取 json 文件 df = spark.read.option("multiline", True).json("/tmp/data.json")但是，当运行 display(df) 时，我只看到一条记录。请问如何显示所有记录？
@gregeal 我将编辑我的问题解释原因
我正在从数据库中提取数据，无法更改源中正确的 JSON 格式，如何通过 json.loads 将其转换为数组以便解析？感谢您的帮助

【解决方案2】：

您正在尝试将 JSON 字符串指定为数据的路径 - 它不起作用。 .json() 函数要么接受文件路径，要么接受字符串的 RDD。要创建 RDD，请使用以下代码：

rdd = sc.parallelize([data1])
df = spark.read.json(rdd)

【讨论】：