Spark宽稀疏DataFrame持久化的最优存储格式及Spark端忽略空值配置方案咨询

嘿，针对你这种超宽、超高稀疏度的Spark DataFrame存储场景，我来给你梳理下最优方案和实操细节：

从你的测试结果（Parquet 55MB vs Avro 15MB）能看出来，常规的列式存储格式默认还是会保留空值的元数据，没法完全利用稀疏特性。针对你的场景，优先级推荐如下：

1. HBase（首推） #

HBase天生就是为稀疏、超宽表设计的：

• 它采用动态列模型，不需要预先定义所有10万+列，每行只存储非空的列（列族+列名+值），完全忽略null值，存储成本能降到最低。
• 对于超宽表的元数据管理压力远小于Parquet/Avro，后者在处理十万级列时，元数据文件本身就会占用不少空间。
• 后续查询时也能直接定位到非空列，效率更高。

2. 转长表后的Parquet/Avro（次选） #

如果不想引入HBase的运维成本，建议先把你的宽表转成长表格式（只保留非空值），再用Parquet或Avro存储：

• 把原来的[row_id, col1, col2, ..., col130000]结构转成[row_id, col_name, col_value]，只保留col_value非空的行。
• 这种方式下，你的100行×13万列数据，非空值仅约13万条，存储量会比你之前测试的15MB再大幅下降，而且Parquet/Avro的压缩和列式优化能进一步节省空间。

针对HBase的写入配置 #

1. 首先引入Spark-HBase连接器（根据你的Spark和HBase版本调整依赖），比如Maven依赖：

<dependency>
  <groupId>org.apache.hadoop.hbase</groupId>
  <artifactId>hbase-spark</artifactId>
  <version>2.4.9</version> <!-- 匹配你的HBase版本 -->
</dependency>

2. 编写代码时，只生成非空列的Put对象：

// Scala示例，Python逻辑类似
import org.apache.hadoop.hbase.{HBaseConfiguration, TableName}
import org.apache.hadoop.hbase.client.Put
import org.apache.hadoop.hbase.io.ImmutableBytesWritable
import org.apache.hadoop.hbase.util.Bytes
import org.apache.spark.sql.functions._

// 假设row_id是每行的唯一标识
val hbaseConf = HBaseConfiguration.create()
hbaseConf.set(Table.NAME, "your_table_name")

val hbaseRDD = wideDf.rdd.map(row => {
  val rowId = Bytes.toBytes(row.getAs[String]("row_id"))
  val put = new Put(rowId)
  // 遍历所有列，只添加非空值
  wideDf.columns.filter(_ != "row_id").foreach(col => {
    val value = row.getAs[Any](col)
    if (value != null) {
      put.addColumn(
        Bytes.toBytes("your_column_family"),
        Bytes.toBytes(col),
        Bytes.toBytes(value.toString) // 根据实际类型调整序列化方式
      )
    }
  })
  (new ImmutableBytesWritable, put)
})

// 写入HBase
hbaseRDD.saveAsNewAPIHadoopDataset(hbaseConf)

针对Parquet/Avro的转长表优化 #

直接把宽表转成非空长表后再存储，以Python为例：

from pyspark.sql.functions import explode, array, struct, lit, col

# 假设row_id是每行的唯一标识
# 构造所有列的struct数组，包含列名和列值
cols_struct = array(*[struct(lit(c).alias("col_name"), col(c).alias("col_value")) for c in df.columns if c != "row_id"])
long_df = df.select("row_id", explode(cols_struct).alias("col_info")) \
            .select("row_id", "col_info.col_name", "col_info.col_value") \
            .filter(col("col_value").isNotNull())

# 写入Parquet（Avro只需把parquet改成avro，配置对应压缩）
long_df.write.mode("overwrite") \
      .option("compression", "snappy") \
      .parquet("/path/to/your/storage")

额外配置：

• Parquet：开启spark.sql.parquet.enable.dictionary（默认true），配合snappy压缩，进一步优化存储。
• Avro：设置avro.compression.codec=snappy，提升压缩效率。

你之前直接写入Parquet/Avro时，即使是空值，格式还是会为每个空列保留元数据和空值标记，所以空间占用高。转长表或者用HBase，本质都是只存储有效非空数据，这才是解决你问题的核心。

内容的提问来源于stack exchange，提问作者py-r