使用 Pyspark 从数组中读取 JSON 项？

Question

我在从 Databricks 中的 Cosmos DB 读取项目时遇到了一些问题，它似乎将 JSON 作为字符串值读取，并且在将数据从中获取到列时遇到了一些问题。

我有一个名为 ProductRanges 的列，其中连续包含以下值：

[   {
        "name": "Red",
        "min": 0,
        "max": 99,
        "value": "Order More"
    },
    {
        "name": "Amber",
        "min": 100,
        "max": 499,
        "value": "Stock OK"
    },
    {
        "name": "Green",
        "min": 500,
        "max": 1000000,
        "value": "Overstocked"
    }
]

在 Cosmos DB 中，JSON 文档是有效的，在导入数据时，数据框中的数据类型是字符串，而不是我期望的 JSON 对象/结构。

我希望能够计算“名称”出现的次数并遍历它们以获取最小值、最大值和值项，因为我们可以拥有的范围数可能超过 3。我已经在 stackoverflow 和其他地方发表了几篇文章，但一直停留在格式上。我尝试使用爆炸并读取基于列值的架构，但它确实说“在有效文档中”，认为这可能是由于 Pyspark 在开始和结束时需要 {}，但甚至将其连接在来自 cosmos db 的 SQL 查询仍然以字符串的数据类型结束。

任何指针将不胜感激

Answer 1

我看到您从 Azure CosmosDB 检索 JSON 文档并将它们转换为 PySpark DataFrame，但嵌套的 JSON 文档或数组无法按您的预期转换为 DataFrame 列中的 JSON 对象，因为没有定义 JSON 类型在pyspark.sql.types 模块中，如下所示。

我搜索了一个文档PySpark: Convert JSON String Column to Array of Object (StructType) in Data Frame，它是您当前案例的合适解决方案，甚至与您想要的相同，而我正在尝试解决它。​​

上面的文档展示了如何使用ArrayType、StructType、StructField 和其他基本 PySpark 数据类型将列中的 JSON 字符串转换为可以在 PySpark 中通过定义列模式更容易处理的组合数据类型和一个 UDF。

这里是示例代码的摘要。希望对您有所帮助。

source = [{"attr_1": 1, "attr_2": "[{\"a\":1,\"b\":1},{\"a\":2,\"b\":2}]"}, {"attr_1": 2, "attr_2": "[{\"a\":3,\"b\":3},{\"a\":4,\"b\":4}]"}]

通过 sqlContext 将 JSON 读入数据框。输出是：

+------+--------------------+

|attr_1|              attr_2|

+------+--------------------+

|     1|[{"a":1,"b":1},{"...|

|     2|[{"a":3,"b":3},{"...|

+------+--------------------+


root
  |-- attr_1: long (nullable = true)
  |-- attr_2: string (nullable = true)

然后，通过定义列架构和 UDF 来转换 attr_2 列。

# Function to convert JSON array string to a list
import json

def parse_json(array_str):
    json_obj = json.loads(array_str)
    for item in json_obj:
        yield (item["a"], item["b"])

# Define the schema
from pyspark.sql.types import ArrayType, IntegerType, StructType, StructField

json_schema = ArrayType(StructType([StructField('a', IntegerType(
), nullable=False), StructField('b', IntegerType(), nullable=False)]))

# Define udf
from pyspark.sql.functions import udf

udf_parse_json = udf(lambda str: parse_json(str), json_schema)

# Generate a new data frame with the expected schema

df_new = df.select(df.attr_1, udf_parse_json(df.attr_2).alias("attr_2"))
df_new.show()
df_new.printSchema()

输出如下：

+------+--------------+

|attr_1|        attr_2|

+------+--------------+

|     1|[[1,1], [2,2]]|

|     2|[[3,3], [4,4]]|

+------+--------------+


root
  |-- attr_1: long (nullable = true)
  |-- attr_2: array (nullable = true)
  |    |-- element: struct (containsNull = true)
  |    |    |-- a: integer (nullable = false)
  |    |    |-- b: integer (nullable = false)