PySpark 解析 JSON 属性为表格列方法

本教程详细介绍了如何使用 PySpark 将 Oracle REST API 返回的 JSON 数组数据（其中属性名和属性值以键值对形式存在）转换为结构化的表格格式。通过 PySpark 读取 JSON 数据并结合 Spark SQL 的 MAX(CASE WHEN ...) 语句，实现将动态属性名称（如 'LOG_ID' 和 'BUSINESS_UNIT'）透视为独立的列，从而方便数据分析和处理。

在数据集成和处理过程中，我们经常会遇到来自 REST API 的响应数据，其结构可能并非传统的行列表格形式。例如，某些 API 会以键值对数组的形式返回数据，其中每个对象包含一个属性名（attributeName）和对应的属性值（attributeValue）。当需要将这些动态的属性名转换为固定的列，并将其对应的属性值填充到这些列中时，传统的转换方法可能不够灵活。本教程将展示如何利用 PySpark 的强大能力，特别是结合 Spark SQL，高效地实现这种数据透视操作。

问题描述

假设我们从 Oracle REST API 获得以下 JSON 响应数据：

[
    {
        "attributeId": 300000000227671,
        "attributeName": "BUSINESS_UNIT",
        "attributeType": "Number",
        "attributeValue": "300000207138371",
        "timeBuildingBlockId": 300000300319699,
        "timeBuildingBlockVersion": 1
    },
    {
        "attributeId": 300000000226689,
        "attributeName": "LOG_ID",
        "attributeType": "Number",
        "attributeValue": "300000001228038",
        "timeBuildingBlockId": 300000300319699,
        "timeBuildingBlockVersion": 1
    }
]

我们的目标是将 attributeName 为 'LOG_ID' 和 'BUSINESS_UNIT' 的 attributeValue 提取出来，并将其转换为以下表格形式：

LOG_ID	BUSINESS_UNIT
300000001228038	300000207138371

解决方案：使用 PySpark 和 Spark SQL

PySpark 提供了强大的数据处理能力，结合 Spark SQL，可以非常灵活地处理这种数据透视场景。核心思路是先将 JSON 数据加载到 DataFrame 中，然后利用 Spark SQL 的条件聚合函数（CASE WHEN 和 MAX）实现透视。

步骤一：加载 JSON 数据到 DataFrame

首先，我们需要将 JSON 响应数据加载到 PySpark DataFrame 中。假设 json_data 是包含上述 JSON 字符串的变量。

from pyspark.sql import SparkSession

# 初始化 SparkSession
spark = SparkSession.builder.appName("JsonPivotTutorial").getOrCreate()
sc = spark.sparkContext

# 模拟 JSON 数据，实际应用中可能是从文件或API响应获取
json_data = """
[
    {
        "attributeId": 300000000227671,
        "attributeName": "BUSINESS_UNIT",
        "attributeType": "Number",
        "attributeValue": "300000207138371",
        "timeBuildingBlockId": 300000300319699,
        "timeBuildingBlockVersion": 1
    },
    {
        "attributeId": 300000000226689,
        "attributeName": "LOG_ID",
        "attributeType": "Number",
        "attributeValue": "300000001228038",
        "timeBuildingBlockId": 300000300319699,
        "timeBuildingBlockVersion": 1
    }
]
"""

# 将 JSON 字符串转换为 RDD 并读取为 DataFrame
# 注意：如果 json_data 是一个列表，可以直接使用 spark.createDataFrame()
# 但如果是一个多行 JSON 字符串，或者需要更灵活地处理，spark.read.json(sc.parallelize([json_data])) 是一个有效方法
df = spark.read.json(sc.parallelize([json_data]))

# 查看原始 DataFrame 结构
df.printSchema()
df.show(truncate=False)

执行上述代码后，df 将包含解析后的 JSON 数据，每行对应 JSON 数组中的一个对象。

步骤二：创建临时视图

为了方便使用 Spark SQL 进行查询，我们将 DataFrame 注册为一个临时视图（Temporary View）。

df.createOrReplaceTempView("myTable")

现在，我们可以像操作传统数据库表一样，通过 SQL 语句查询 myTable。

步骤三：使用 Spark SQL 进行数据透视

透视的核心在于使用 CASE WHEN 语句根据 attributeName 的值选择对应的 attributeValue，并通过聚合函数（如 MAX）将每个组中的非空值提取出来。由于我们希望将所有相关属性（例如 LOG_ID 和 BUSINESS_UNIT，它们共享相同的 timeBuildingBlockId 和 timeBuildingBlockVersion）聚合到一行，因此需要对这些共享字段进行隐式分组。

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
    GROUP BY timeBuildingBlockId, timeBuildingBlockVersion -- 根据业务逻辑分组，确保同一逻辑实体的数据聚合到一行
""")

result.show()

SQL 逻辑解释：

• CASE WHEN attributeName = 'LOG_ID' THEN attributeValue END: 这部分逻辑会检查 attributeName 是否为 'LOG_ID'。如果是，则返回对应的 attributeValue；否则返回 NULL。
• MAX(...) AS LOG_ID: 由于每个 attributeName 对应的 attributeValue 在原始数据中只出现一次（对于特定的逻辑实体），所以 MAX 函数会从 CASE WHEN 表达式生成的多个 NULL 值和一个非 NULL 值中选择那个非 NULL 的 attributeValue。这有效地将特定属性的 attributeValue 提升为新的列。
• GROUP BY timeBuildingBlockId, timeBuildingBlockVersion: 这一步至关重要。原始 JSON 数据中，LOG_ID 和 BUSINESS_UNIT 属于同一个逻辑实体，它们共享相同的 timeBuildingBlockId 和 timeBuildingBlockVersion。通过对这些字段进行分组，我们可以确保属于同一逻辑实体（即同一组）的所有属性值被聚合到同一行中。如果没有 GROUP BY，或者分组字段选择不当，可能会导致结果不正确（例如，所有属性聚合到一行，或者数据被错误地分割）。

输出结果：

+---------------+-------------------+
|LOG_ID         |BUSINESS_UNIT      |
+---------------+-------------------+
|300000001228038|300000207138371|
+---------------+-------------------+

这正是我们期望的透视结果。

注意事项与总结

1. 动态列处理： 上述方法适用于列名（LOG_ID, BUSINESS_UNIT）已知的情况。如果 attributeName 的种类是动态变化的，并且需要在运行时确定列名，则需要结合 PySpark 的 DataFrame API 中的 pivot 函数，或者在 Spark SQL 中使用动态 SQL 生成技术。然而，对于固定的少量列，CASE WHEN 语句更直接和高效。
2. 聚合函数选择： 除了 MAX，也可以根据实际需求选择其他聚合函数，如 MIN、SUM、AVG 等。但对于这种将单个值提升为列的场景，MAX（或 MIN）是最常见的选择，因为它会忽略 NULL 值并返回唯一的非 NULL 值。
3. 分组键的重要性： GROUP BY 子句的选择至关重要。它决定了哪些原始行的数据会被聚合成新的一行。在上述示例中，timeBuildingBlockId 和 timeBuildingBlockVersion 共同标识了一个唯一的业务实体，因此它们是理想的分组键。务必根据您的数据模型和业务需求来确定正确的分组键。
4. 性能考量： 对于非常大的数据集，Spark SQL 能够有效地并行处理数据。然而，过多的 CASE WHEN 表达式或过于复杂的分组逻辑可能会影响性能。在实际应用中，应根据数据量和集群资源进行调优。

通过 PySpark 和 Spark SQL 的结合，我们可以灵活高效地处理各种复杂的数据转换需求，将非结构化或半结构化的 JSON 数据转换为易于分析的表格格式。