本文详解如何在 PySpark 中根据 n_relevant 字段动态对固定长度数组（如 300 元素）执行三种策略：全量截取（
本文详解如何在 pyspark 中根据 `n_relevant` 字段动态对固定长度数组（如 300 元素）执行三种策略：全量截取（
在实际数据工程场景中（如 Azure Databricks），常需对长数组进行动态裁剪以满足下游系统约束（例如 sink 要求数组长度 ≤ 100）。原始需求并非简单切片，而是依据辅助列 n_relevant 自适应选择策略：
- 若
n_relevant ：直接取前 <code>n_relevant个元素； - 若
100 ≤ n_relevant ：在前 <code>n_relevant个元素中，按ceil(n_relevant / 100)步长采样（即模运算）； - 若
n_relevant ≥ 300：统一按模 3 采样（等价于i % 3 == 0，索引从 0 开始）。
但 PySpark 原生函数（如 transform、array_remove）无法将列值传入 lambda，导致动态模数难以实现。核心解法是分层组合 slice + filter + when/otherwise，并巧妙利用 filter 的双参数签名（lambda elem, index:）访问元素索引。
以下为完整可运行示例（适配 Spark 3.4+）：
from pyspark.sql import functions as F
from pyspark.sql.types import StructType, StructField, ArrayType, IntegerType
# 构造测试数据：array 固定为 [0,1,...,299]，n_relevant 取不同典型值
df = spark.createDataFrame(
[[list(range(300)), 4],
[list(range(300)), 200],
[list(range(300)), 300],
[list(range(300)), 800]],
schema=StructType([
StructField("array", ArrayType(IntegerType())),
StructField("n_relevant", IntegerType())
])
)
# 动态采样逻辑（注意：Spark 中 array 索引从 1 开始，但 filter 的 index 参数从 0 开始！）
df_result = df.withColumn(
"result",
F.when(
F.col("n_relevant") = 100) & (F.col("n_relevant") = 300：统一模 3 采样（索引 0,3,6... → 对应值 array[0],array[3],array[6]...）
F.filter(
F.slice("array", 1, 300), # 安全 slice 至最大长度
lambda _, idx: idx % 3 == 0
)
)
)
display(df_result.select("n_relevant", "result"))
⚠️ 关键注意事项：
-
F.filter(array_col, lambda elem, index:)中的index是 0-based，而F.slice(col, start, length)的start是 1-based，务必区分； - Spark SQL 函数链式调用中，
F.ceil(F.col("n_relevant") / 100)返回的是 Column 类型，可直接用于模运算右侧（Spark 3.4+ 支持）； - 若需严格匹配题设中
ceil(n_relevant/100)的动态步长（如n=150 → mod=2,n=250 → mod=3），上述when分支需进一步拆解，或改用F.expr("filter(slice(array,1,n_relevant), (x,i) -> i % cast(ceil(n_relevant/100) as int) == 0)")； - 对于超复杂逻辑，建议封装为 Pandas UDF（
pandas_udf）或 SQL UDF，兼顾可读性与扩展性； - 生产环境务必添加
n_relevant非负校验（F.when(F.col("n_relevant") ），避免 <code>slice报错。
该方案避免了低效的 collect() 和 Python UDF 序列化开销，在 Spark Catalyst 优化器下可高效执行，是处理“条件数组采样”类问题的标准范式。
