基于Spark实现百万行属性表转宽表的方法咨询

完全可以通过Spark实现这个需求，而且因为你的attr_name唯一值不超过20个（基数很小），Spark的pivot算子能高效处理百万级数据量。

核心实现方法：groupBy + pivot #

Spark DataFrame API提供的pivot方法专门用于行转列场景，结合groupBy按主键分组后，就能快速生成宽表结构。

步骤拆解 #

1. 分组：以object_id作为分组键，确保最终宽表以它为主键。
2. 透视列：指定attr_name作为透视字段，将每个唯一的属性名转为宽表的列。
3. 聚合取值：对value列执行聚合操作（因为每个object_id+attr_name组合应唯一，用first/max等取唯一值即可）。

代码示例 #

Python 版本

from pyspark.sql import SparkSession
from pyspark.sql.functions import first

# 初始化Spark会话
spark = SparkSession.builder.appName("LongToWide").getOrCreate()

# 读取原始数据（支持Parquet/CSV/JDBC等多种数据源）
raw_df = spark.read.parquet("your/raw/data/path")

# 执行行转列
wide_table = raw_df.groupBy("object_id") \
                   .pivot("attr_name") \
                   .agg(first("value"))

# 查看结果或保存
wide_table.show()
wide_table.write.parquet("your/wide/data/save/path", mode="overwrite")

Scala 版本

import org.apache.spark.sql.SparkSession
import org.apache.spark.sql.functions.first

object LongToWideConverter {
  def main(args: Array[String]): Unit = {
    val spark = SparkSession.builder.appName("LongToWide").getOrCreate()
    import spark.implicits._

    // 读取原始数据
    val rawDf = spark.read.parquet("your/raw/data/path")

    // 生成宽表
    val wideTable = rawDf.groupBy("object_id")
                         .pivot("attr_name")
                         .agg(first("value"))

    // 输出或保存
    wideTable.show()
    wideTable.write.parquet("your/wide/data/save/path", mode = "overwrite")
  }
}

关键注意事项 #

• 性能优化：由于attr_name唯一值少，Spark会自动优化pivot的执行计划，无需额外配置即可高效处理百万级数据。如果后续属性数量增加，可提前指定透视列列表（如.pivot("attr_name", List("attr1", "attr2"))）进一步提升性能。
• 重复值处理：若存在同一object_id+attr_name有多条记录的情况，需根据业务逻辑选择聚合函数（比如first取第一条、max取最大值等）。
• 空值填充：宽表中缺失属性值的位置会显示null，可通过wide_table.fillna("默认值")或wide_table.fillna(0)统一填充。

学习资料参考 #

• Spark官方DataFrame API文档：重点查看GroupedData类的pivot方法说明，了解参数细节和优化方式。
• 《Spark权威指南》：书中数据转换章节有行转列的实战案例，涵盖场景分析和代码讲解。

内容的提问来源于stack exchange，提问作者Maxim