使用 PySpark 从 JSON 对象中提取并透视数据-Python教程-PHP中文网

使用 PySpark 从 JSON 对象中提取并透视数据

花韻仙語

发布： 2025-08-19 16:20:28

原创

186人浏览过

使用 pyspark 从 json 对象中提取并透视数据

本文介绍如何使用 PySpark 从包含属性和值的 JSON 对象中提取特定列，并将其透视为所需的格式。通过创建 DataFrame 和使用 Spark SQL，可以灵活地选择和转换 JSON 数据，最终得到结构化的结果。本文提供详细的代码示例，帮助你理解和应用这种方法。

使用 PySpark 从 JSON 对象中提取并透视数据

当需要从 REST API 响应的 JSON 数据中提取特定信息，并将其转换为更易于分析的格式时，可以使用 PySpark。以下步骤展示了如何使用 PySpark 从 JSON 对象中选择特定列，并将其透视为所需的格式，例如将 attributeName 作为列名，attributeValue 作为对应的值。

步骤 1：创建 DataFrame

首先，需要将 JSON 数据加载到 PySpark 的 DataFrame 中。可以使用 spark.read.json() 函数来完成此操作。

from pyspark.sql import SparkSession

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

# 示例 JSON 数据 (假设从 REST API 获取)
json_data = """
[
    {
        "attributeId": 300000000227671,
        "attributeName": "BUSINESS_UNIT",
        "attributeType": "Number",
        "attributeValue": "300000207138371",
        "timeBuildingBlockId": 300000300319699,
        "timeBuildingBlockVersion": 1
    },
    {
        "attributeId": 300000000227689,
        "attributeName": "LDG_ID",
        "attributeType": "Number",
        "attributeValue": "300000001228038",
        "timeBuildingBlockId": 300000300319699,
        "timeBuildingBlockVersion": 1
    }
]
"""

# 将 JSON 数据转换为 DataFrame
df = spark.read.json(spark.sparkContext.parallelize([json_data]))

# 打印 DataFrame 的 Schema，方便理解数据结构
df.printSchema()

# 显示 DataFrame 的内容
df.show()

登录后复制

这段代码首先创建了一个 SparkSession，这是与 Spark 集群交互的入口点。然后，定义了示例 JSON 数据。spark.read.json() 函数读取 JSON 数据，并使用 spark.sparkContext.parallelize() 将其转换为 RDD，以便 Spark 可以并行处理数据。df.printSchema() 和 df.show() 用于查看 DataFrame 的结构和内容，帮助理解数据的组织方式。

步骤 2：创建临时视图

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

df.createOrReplaceTempView("myTable")

登录后复制

createOrReplaceTempView() 函数将 DataFrame 注册为一个名为 myTable 的临时视图。如果已经存在同名的视图，则会替换它。

Krikey AI

查看详情

步骤 3：使用 Spark SQL 进行透视

使用 Spark SQL 的 CASE WHEN 语句可以实现数据的透视。以下代码演示了如何提取 attributeName 为 LOG_ID 和 BUSINESS_UNIT 的 attributeValue，并将它们作为新的列。

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

result.show()

登录后复制

这段 SQL 代码使用 CASE WHEN 语句根据 attributeName 的值来选择对应的 attributeValue。MAX() 函数用于处理可能存在的多个匹配项，确保每个 attributeName 只返回一个值。最终的结果是一个包含 LDG_ID 和 BUSINESS_UNIT 两列的 DataFrame。

完整代码示例

将上述步骤组合在一起，得到完整的 PySpark 代码：

from pyspark.sql import SparkSession

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

# 示例 JSON 数据 (假设从 REST API 获取)
json_data = """
[
    {
        "attributeId": 300000000227671,
        "attributeName": "BUSINESS_UNIT",
        "attributeType": "Number",
        "attributeValue": "300000207138371",
        "timeBuildingBlockId": 300000300319699,
        "timeBuildingBlockVersion": 1
    },
    {
        "attributeId": 300000000227689,
        "attributeName": "LDG_ID",
        "attributeType": "Number",
        "attributeValue": "300000001228038",
        "timeBuildingBlockId": 300000300319699,
        "timeBuildingBlockVersion": 1
    }
]
"""

# 将 JSON 数据转换为 DataFrame
df = spark.read.json(spark.sparkContext.parallelize([json_data]))

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

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

# 显示结果
result.show()

# 停止 SparkSession
spark.stop()

登录后复制

注意事项

数据类型： 确保 attributeValue 的数据类型与你的需求相符。如果需要进行数值计算，可能需要将其转换为数值类型。
错误处理： 在实际应用中，应该添加错误处理机制，例如检查 JSON 数据是否有效，以及处理 attributeName 不存在的情况。
性能优化： 对于大型数据集，可以考虑使用 Spark 的分区和广播等功能来优化性能。

总结

通过使用 PySpark 和 Spark SQL，可以灵活地从 JSON 对象中提取和转换数据。这种方法适用于处理来自 REST API 或其他数据源的半结构化数据，并将其转换为结构化的格式，以便进行进一步的分析和处理。

以上就是使用 PySpark 从 JSON 对象中提取并透视数据的详细内容，更多请关注php中文网其它相关文章！