本文介绍如何使用 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 的临时视图。如果已经存在同名的视图，则会替换它。
步骤 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 或其他数据源的半结构化数据，并将其转换为结构化的格式，以便进行进一步的分析和处理。
