本文介绍如何在 PySpark 中仅对指定列（如 function_txt、item_txt、value_txt）在相同分组（如 diag_sid 和 vehicle_id）内进行随机重排，保持其他列和分组结构不变。
本文介绍如何在 pyspark 中仅对指定列（如 `function_txt`、`item_txt`、`value_txt`）在相同分组（如 `diag_sid` 和 `vehicle_id`）内进行随机重排，保持其他列和分组结构不变。
在数据脱敏、模型训练前的数据增强或测试场景中，常需对某些敏感字段（如诊断文本、编码值）在逻辑分组内进行“内部洗牌”——即保持每组的记录数和分组键不变，但打乱该组内特定列的行间对应关系。这不同于全局重排序（df.orderBy(F.rand())），也不同于单行内字段置换，而是跨行重分配指定列的值，同时确保其他列（如 diag_sid、vehicle_id、source）严格对齐原始分组。
PySpark 本身不提供直接的“列级 shuffle”函数，但可通过 结构化列 + 窗口函数 + 随机排序 的组合方式高效实现。核心思路如下：
- 构造随机结构体（Struct）：将待打乱的列（如 function_txt, item_txt, value_txt）与一个随机整数（rand_int）打包为一个 struct 类型列；
- 定义分组窗口（Window）：以 diag_sid 和 vehicle_id 为 partitionBy，以结构体中的 rand_int 为 orderBy 字段；
- 重排序并展开：利用 row_number() 获取新顺序索引，再通过 collect_list() + element_at() 或更优的 结构体广播重映射 实现列值重分配——但注意：原答案中未完成最终展开步骤，实际需补充关键一步：用窗口内随机序号对结构体列表做索引重取。
以下是生产就绪的完整实现（兼容 Spark 3.0+，无需 UDF，纯 SQL 函数）：
from pyspark.sql import functions as F
from pyspark.sql.window import Window
# 假设 df 已存在，包含列：diag_sid, vehicle_id, source, function_txt, item_txt, value_txt, date
shuffle_cols = ["function_txt", "item_txt", "value_txt"]
# 步骤1：为每行生成独立随机整数（确保同一分组内可比较）
df_with_rand = df.withColumn("rand_seed", F.expr("floor(rand() * 1000000)").cast("int"))
# 步骤2：将待打乱列 + 随机种子打包为 struct，并保留原始分组键
struct_col = F.struct("rand_seed", *[F.col(c) for c in shuffle_cols])
df_struct = df_with_rand.withColumn("shuffled_block", struct_col).drop(*shuffle_cols, "rand_seed")
# 步骤3：定义窗口 —— 按分组键分区，按随机种子排序
w = Window.partitionBy("diag_sid", "vehicle_id").orderBy("shuffled_block.rand_seed")
# 步骤4：收集本组所有 shuffled_block，并随机打乱其顺序（关键！）
# 使用 collect_list + shuffle（Spark 3.4+ 支持 array_shuffle；旧版可用 sort_array + rand）
df_collected = df_struct.withColumn(
"group_blocks",
F.collect_list("shuffled_block").over(w)
).withColumn(
"shuffled_blocks",
F.expr("shuffle(group_blocks)") # Spark 3.4+；若版本较低，改用：
# F.sort_array(F.col("group_blocks"), asc=F.rand() > 0.5)
).drop("shuffled_block", "group_blocks")
# 步骤5：为每行分配新索引（0-based），用于从 shuffled_blocks 中取值
df_indexed = df_collected.withColumn(
"idx",
F.row_number().over(w) - 1 # 转为 0-based 索引
)
# 步骤6：按 idx 取出打乱后的 struct，并展开各字段
result_df = df_indexed.select(
"*",
F.col("shuffled_blocks")[F.col("idx")].alias("reassigned")
).select(
"diag_sid",
"vehicle_id",
"source", # 保持不变的列
"date", # 保持不变的列
F.col("reassigned.function_txt").alias("function_txt"),
F.col("reassigned.item_txt").alias("item_txt"),
F.col("reassigned.value_txt").alias("value_txt")
)✅ 关键优势：
- 完全基于 Catalyst 优化器，无 UDF 开销；
- 保证每个分组内 shuffle_cols 的值被完全重排（无重复、无遗漏）；
- 其他列（source, date 等）不受影响，逻辑一致性得以维持。
⚠️ 注意事项：
- 若分组内记录数极少（如仅 1 行），打乱效果不可见；
- shuffle() 函数在 Spark < 3.4 中不可用，请替换为 sort_array(col, rand() > 0.5)；
- 随机性依赖 rand()，如需可复现结果，应设置 spark.sql.adaptive.enabled=false 并使用固定种子（rand(42)）；
- 大分组（万级行）下 collect_list 可能触发内存压力，建议监控 spark.sql.adaptive.enabled 和 spark.sql.adaptive.coalescePartitions.enabled。
通过该方案，你即可安全、高效地实现“组内列值洗牌”，满足数据扰动、隐私保护或测试覆盖等典型工程需求。
