您可以使用sequence函数创建从1到size的序列数组，然后将其分解：

import org.apache.spark.sql.functions._
val df = Seq((1,4), (2,2)).toDF("id", "size")
df
  .withColumn("size", explode(sequence(lit(1), col("size"))))
  .show(false)

输出结果为：

+---+----+
|id |size|
+---+----+
|1  |1   |
|1  |2   |
|1  |3   |
|1  |4   |
|2  |1   |
|2  |2   |
+---+----+

您可以使用first use sequence函数创建从1到size的序列，然后将其分解。

val df = input.withColumn("seq", sequence(lit(1), $"size"))
df.show()
+---+----+------------+
| id|size|         seq|
+---+----+------------+
|  1|   4|[1, 2, 3, 4]|
|  2|   2|      [1, 2]|
+---+----+------------+

df.withColumn("size", explode($"seq")).drop("seq").show()
+---+----+
| id|size|
+---+----+
|  1|   1|
|  1|   2|
|  1|   3|
|  1|   4|
|  2|   1|
|  2|   2|
+---+----+

为了完整起见，不使用显式explode：

val df = Seq((1,4),(2,2)).toDF("id","size")
val n = spark.range(1,1000).toDF("n")
df.join(broadcast(n),col("size") >= col("n")).show(false)
+---+----+---+                                                                  
|id |size|n  |
+---+----+---+
|1  |4   |1  |
|1  |4   |2  |
|1  |4   |3  |
|1  |4   |4  |
|2  |2   |1  |
|2  |2   |2  |
+---+----+---+

...并根据需要删除/重命名列。

你可以使用Seq.range将size列转换成一个递增序列，然后分解数组，如下所示：

import spark.implicits._
import org.apache.spark.sql.functions.{explode, col}

// Original dataframe
val df = Seq((1,4), (2,2)).toDF("id", "size")

// Mapping over this dataframe: turning each row into (idx, array)
val df_with_array = df
  .map(row => {
    (row.getInt(0), Seq.range(1, row.getInt(1) + 1)) 
  })
  .toDF("id", "array")
  .select(col("id"), explode(col("array")))

output.show()
+---+---+
| id|col|
+---+---+
|  1|  1|
|  1|  2|
|  1|  3|
|  1|  4|
|  2|  1|
|  2|  2|
+---+---+