如何确保值正确映射至Delta Table列？PySpark数据处理需求问询

方法一：固定列提取（适用于已知所有键的场景） #

直接针对目标列（a、b、c、d、e、j、l）进行提取，步骤如下：

1. 初始化SparkSession并读取Delta表

from pyspark.sql import SparkSession
from pyspark.sql.functions import str_to_map, col

# 创建SparkSession
spark = SparkSession.builder.appName("SplitValuesToColumns").getOrCreate()

# 读取源Delta表
source_df = spark.read.format("delta").load("/path/to/your/table1")

2. 将Values列转换为键值对Map
   使用str_to_map函数，按&拆分键值对，按=拆分键和值：

df_with_map = source_df.withColumn("values_map", str_to_map(col("values"), "&", "="))

3. 提取Map中的键作为单独列并转换类型
   将每个键对应的值转为整数类型，匹配目标格式：

final_df = df_with_map.select(
    col("ID"),
    col("values_map").getItem("a").cast("integer").alias("a"),
    col("values_map").getItem("b").cast("integer").alias("b"),
    col("values_map").getItem("c").cast("integer").alias("c"),
    col("values_map").getItem("d").cast("integer").alias("d"),
    col("values_map").getItem("e").cast("integer").alias("e"),
    col("values_map").getItem("j").cast("integer").alias("j"),
    col("values_map").getItem("l").cast("integer").alias("l")
)

4. 写入新的Delta表

# 写入目标Delta表，可根据需求调整mode（如append、ignore等）
final_df.write.format("delta").mode("overwrite").save("/path/to/your/new_delta_table")

方法二：动态列提取（适用于键不固定的场景） #

如果Values列的键可能变化，可自动识别所有唯一键并生成对应列：

# 步骤1-2同方法一，获取带Map列的DataFrame后：

# 提取所有唯一的键
all_unique_keys = df_with_map.select("values_map").rdd.flatMap(lambda row: row[0].keys()).distinct().collect()

# 动态生成列选择表达式
select_expressions = [col("ID")] + [
    col("values_map").getItem(key).cast("integer").alias(key) 
    for key in all_unique_keys
]

# 生成最终DataFrame
final_df = df_with_map.select(*select_expressions)

# 写入Delta表（同方法一步骤4）
final_df.write.format("delta").mode("overwrite").save("/path/to/your/new_delta_table")

关键说明 #

• str_to_map函数是PySpark处理这种键值对字符串的高效方式，避免了多次split和explode的复杂操作。
• 转换为integer类型是因为样本值均为整数，若存在其他类型可调整cast参数（如string、double）。
• 写入Delta表时，mode参数根据实际需求选择：overwrite覆盖原表，append追加数据，ignore忽略重复写入。

内容的提问来源于stack exchange，提问作者boring-coder