Scala 将窗口分区数据集传递给 UDF

Question

我有一个如下所示的数据框，

Id1	Id2	Id3	TaskId	TaskName	index
1	11	bc123-234	dfr3ws-45d	randomName1	1
1	11	bc123-234	er98d3-lkj	randomName2	2
1	11	bc123-234	hu77d9-mnb	randomName3	3
1	11	bc123-234	xc33d5-rew	deployhere4	4
1	11	xre43-876	dfr3ws-45d	randomName1	1
1	11	xre43-876	er98d3-lkj	deployhere2	2
1	11	xre43-876	hu77d9-mnb	randomName3	3
1	11	xre43-876	xc33d5-rew	randomName4	4

我使用 Id3 和 Id2 对数据进行了分区，并添加了 row_number。

我需要执行以下条件：

TaskId "hu77d9-mnb" 应该出现在包含 deploy 的任务名称之前。如上表所示，名称将是随机的，我需要读取分区中的每个名称并查看其中包含部署的名称。

如果部署 taskName 索引大于 taskID 索引，则将该值标记为 1，否则标记为 0。

我需要像这样获得决赛桌：

Id1	Id2	Id3	TaskId	TaskName	index	result
1	11	bc123-234	dfr3ws-45d	randomName1	1	1
1	11	bc123-234	er98d3-lkj	randomName2	2	1
1	11	bc123-234	hu77d9-mnb	randomName3	3	1
1	11	bc123-234	xc33d5-rew	deployhere4	4	1
1	11	xre43-876	dfr3ws-45d	randomName1	1	0
1	11	xre43-876	er98d3-lkj	deployhere2	2	0
1	11	xre43-876	hu77d9-mnb	randomName3	3	0
1	11	xre43-876	xc33d5-rew	randomName4	4	0

我被困在这个地方，如何将分区数据传递给 UDF（或其他函数，如 UDAF）并执行此任务。任何建议都会有所帮助。感谢您的宝贵时间。

Answer 1

“deploy”行的索引和特定行的索引（“hu77d9-mnb”）可以通过Window“first”函数分配给每一行，然后进行比较：

val df = Seq(
  (1, 11, "bc123-234", "dfr3ws-45d", "randomName1", 1),
  (1, 11, "bc123-234", "er98d3-lkj", "randomName2", 2),
  (1, 11, "bc123-234", "hu77d9-mnb", "randomName3", 3),
  (1, 11, "bc123-234", "xc33d5-rew", "deployhere4", 4),
  (1, 11, "xre43-876", "dfr3ws-45d", "randomName1", 1),
  (1, 11, "xre43-876", "er98d3-lkj", "deployhere2", 2),
  (1, 11, "xre43-876", "hu77d9-mnb", "randomName3", 3),
  (1, 11, "xre43-876", "xc33d5-rew", "randomName4", 4)
).toDF("Id1", "Id2", "Id3", "TaskID", "TaskName", "index")

val specificTaskId = "hu77d9-mnb"
val idsWindow = Window.partitionBy("Id1", "Id2", "Id3")

df.withColumn("deployIndex",
  first(
    when(instr($"TaskName", "deploy") > 0, $"index").otherwise(null),
    true)
    .over(idsWindow))

  .withColumn("specificTaskIdIndex",
    first(
      when($"TaskID" === lit(specificTaskId), $"index").otherwise(null),
      true)
      .over(idsWindow))

  .withColumn("result",
    when($"specificTaskIdIndex" > $"deployIndex", 0).otherwise(1)
  )

输出（必须删除“deployIndex”和“specificTaskIdIndex”列）：

+---+---+---------+----------+-----------+-----+-----------+-------------------+------+
|Id1|Id2|Id3      |TaskID    |TaskName   |index|deployIndex|specificTaskIdIndex|result|
+---+---+---------+----------+-----------+-----+-----------+-------------------+------+
|1  |11 |bc123-234|dfr3ws-45d|randomName1|1    |4          |3                  |1     |
|1  |11 |bc123-234|er98d3-lkj|randomName2|2    |4          |3                  |1     |
|1  |11 |bc123-234|hu77d9-mnb|randomName3|3    |4          |3                  |1     |
|1  |11 |bc123-234|xc33d5-rew|deployhere4|4    |4          |3                  |1     |
|1  |11 |xre43-876|dfr3ws-45d|randomName1|1    |2          |3                  |0     |
|1  |11 |xre43-876|er98d3-lkj|deployhere2|2    |2          |3                  |0     |
|1  |11 |xre43-876|hu77d9-mnb|randomName3|3    |2          |3                  |0     |
|1  |11 |xre43-876|xc33d5-rew|randomName4|4    |2          |3                  |0     |
+---+---+---------+----------+-----------+-----+-----------+-------------------+------+