Scala实现Spark扁平DataFrame转结构化格式适配DynamoDB

核心思路 #

通过解析扁平化列名的层级结构（如Item.column1.S），动态构建嵌套的Struct类型，将分散的列聚合为符合DynamoDB格式的嵌套结构。这种方式无需硬编码列名，扩展性强且性能最优。

Python 实现代码 #

from pyspark.sql.functions import struct, col

# 获取当前DataFrame的所有列名
all_cols = df.columns

# 解析列名，构建每个column对应的Struct（包含S字段）
column_struct_map = {}
for col_name in all_cols:
    parts = col_name.split('.')
    # 筛选符合"Item.{column}.S"格式的列
    if len(parts) == 3 and parts[0] == 'Item' and parts[2] == 'S':
        column_key = parts[1]
        # 为每个column创建包含S字段的Struct
        column_struct_map[column_key] = struct(col(col_name).alias('S'))

# 构建最外层的Item Struct，包含所有column的Struct
item_struct = struct(*[cs.alias(name) for name, cs in column_struct_map.items()]).alias('Item')

# 生成目标DataFrame
result_df = df.select(item_struct)

Scala 实现代码 #

import org.apache.spark.sql.functions.{struct, col}

// 获取当前DataFrame的所有列名
val allCols = df.columns

// 解析列名，构建每个column对应的Struct（包含S字段）
val columnStructMap = allCols.flatMap { colName =>
    val parts = colName.split("\\.")
    // 筛选符合"Item.{column}.S"格式的列
    if (parts.length == 3 && parts(0) == "Item" && parts(2) == "S") {
        Some(parts(1) -> struct(col(colName).alias("S")))
    } else {
        None
    }
}.toMap

// 构建最外层的Item Struct，包含所有column的Struct
val itemStruct = struct(columnStructMap.map { case (name, cs) => cs.alias(name) }.toSeq: _*).alias("Item")

// 生成目标DataFrame
val resultDF = df.select(itemStruct)

方案优势 #

1. 动态适配：自动识别所有符合格式的列，无需手动指定列名，列数量变化时无需修改代码
2. 性能最优：完全使用Spark内置函数实现，避免UDF带来的性能开销
3. 结构精准：严格匹配DynamoDB导出的嵌套Struct格式，确保数据格式一致性

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