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