当前位置：首页 > 文章列表 > 文章 > python教程 > PySpark提取JSON并透视数据方法

PySpark提取JSON并透视数据方法

2025-08-27 12:36:38 0浏览收藏

在IT行业这个发展更新速度很快的行业，只有不停止的学习，才不会被行业所淘汰。如果你是文章学习者，那么本文《PySpark提取JSON数据并透视方法》就很适合你！本篇内容主要包括##content_title##，希望对大家的知识积累有所帮助，助力实战开发！

使用 PySpark 从 JSON 对象中选择并透视数据

本文档介绍了如何使用 PySpark 从包含属性和值的 JSON 对象中提取特定列，并将其透视为所需格式。通过创建 DataFrame 和使用 Spark SQL，我们可以灵活地选择和转换数据，最终得到以指定属性名作为列名的结果。本文提供详细步骤和示例代码，帮助你轻松完成数据提取和转换任务。

使用 PySpark 处理 JSON 数据并进行透视

在数据处理中，经常需要从 JSON 数据中提取特定字段，并将其转换为更易于分析的格式。当 JSON 数据包含具有属性和值的对象数组时，例如 Oracle REST API 的响应，我们可以使用 PySpark 来选择所需的列，并将其透视为以属性名作为列名的形式。

以下是如何使用 PySpark 实现此目标的步骤：

1. 创建 DataFrame

首先，你需要使用 JSON 数据创建一个 DataFrame。假设你已经将 JSON 数据存储在变量 json_data 中，可以使用以下代码创建 DataFrame：

from pyspark.sql import SparkSession

# 创建 SparkSession
spark = SparkSession.builder.appName("JSONPivot").getOrCreate()

df = spark.read.json(spark.sparkContext.parallelize([json_data]))

# 示例 JSON 数据 (替换为你实际的数据)
json_data = """
[
    {
        "attributeId": 300000000227671,
        "attributeName": "BUSINESS_UNIT",
        "attributeType": "Number",
        "attributeValue": "300000207138371",
        "timeBuildingBlockId": 300000300319699,
        "timeBuildingBlockVersion": 1
    },
    {
        "attributeId": 300000000227689,
        "attributeName": "LOG_ID",
        "attributeType": "Number",
        "attributeValue": "300000001228038",
        "timeBuildingBlockId": 300000300319699,
        "timeBuildingBlockVersion": 1
    }
]
"""

df = spark.read.json(spark.sparkContext.parallelize([json_data]))

df.printSchema()
df.show()

这段代码首先创建了一个 SparkSession，这是与 Spark 集群交互的入口点。然后，它使用 spark.read.json() 方法从 json_data 读取 JSON 数据，并将其转换为 DataFrame。spark.sparkContext.parallelize([json_data]) 用于将 JSON 数据转换为 RDD，然后 spark.read.json() 可以从 RDD 读取数据。 df.printSchema() 打印 DataFrame 的结构，df.show() 显示 DataFrame 的内容。

2. 创建临时视图

为了能够使用 Spark SQL 查询 DataFrame，需要创建一个临时视图：

df.createOrReplaceTempView("myTable")

这将创建一个名为 "myTable" 的临时视图，你可以使用 Spark SQL 查询它。

3. 使用 Spark SQL 进行透视

现在，可以使用 Spark SQL 查询临时视图，以提取所需的列并进行透视。以下是一个示例查询，用于提取 "LOG_ID" 和 "BUSINESS_UNIT" 的 attributeValue：

result = spark.sql("""
    SELECT
        MAX(CASE WHEN attributeName = 'LOG_ID' THEN attributeValue END) AS LOG_ID,
        MAX(CASE WHEN attributeName = 'BUSINESS_UNIT' THEN attributeValue END) AS BUSINESS_UNIT
    FROM myTable
""")

result.show()

这个 SQL 查询使用 CASE WHEN 语句来根据 attributeName 的值选择相应的 attributeValue。 MAX() 函数用于处理可能存在多个具有相同 attributeName 的情况，并确保每个属性只有一个值。AS 关键字用于为结果列指定别名。

完整代码示例

from pyspark.sql import SparkSession

# 创建 SparkSession
spark = SparkSession.builder.appName("JSONPivot").getOrCreate()

# 示例 JSON 数据 (替换为你实际的数据)
json_data = """
[
    {
        "attributeId": 300000000227671,
        "attributeName": "BUSINESS_UNIT",
        "attributeType": "Number",
        "attributeValue": "300000207138371",
        "timeBuildingBlockId": 300000300319699,
        "timeBuildingBlockVersion": 1
    },
    {
        "attributeId": 300000000227689,
        "attributeName": "LOG_ID",
        "attributeType": "Number",
        "attributeValue": "300000001228038",
        "timeBuildingBlockId": 300000300319699,
        "timeBuildingBlockVersion": 1
    }
]
"""

# 创建 DataFrame
df = spark.read.json(spark.sparkContext.parallelize([json_data]))

# 创建临时视图
df.createOrReplaceTempView("myTable")

# 使用 Spark SQL 进行透视
result = spark.sql("""
    SELECT
        MAX(CASE WHEN attributeName = 'LOG_ID' THEN attributeValue END) AS LOG_ID,
        MAX(CASE WHEN attributeName = 'BUSINESS_UNIT' THEN attributeValue END) AS BUSINESS_UNIT
    FROM myTable
""")

# 显示结果
result.show()

# 停止 SparkSession
spark.stop()

注意事项