java - Spark : Complex operation with dataframes

Question

我的输入数据集采用以下格式:

+---+--------+----------+
| id|   refId| timestamp|
+---+--------+----------+
|  1|    null|1548944642|
|  1|29950529|1548937685|
|  2|27510720|1548944885|
|  2|27510720|1548943617|
+---+--------+----------+

需要添加一个新列session，并包含以下转换逻辑:

1. 如果refId 为 null，则 session 值为 true。
   
2. 如果id 和 refId 唯一，则 session 值为 true。
3. 如果id 和 refId 不唯一并且“时间戳大于前一行，则 session 值为 true”。此外，时间戳之间的差异应 >60。

+---+--------+-------+----------+
| id|   refId|session| timestamp|
+---+--------+-------+----------+
|  1|    null|   true|1548944642|
|  1|29950529|   true|1548937685|
|  2|27510720|  false|1548943617|
|  2|27510720|   true|1548944885|
+---+--------+-------+----------+

我可以分别满足第 1 和第 3 个条件，但不能满足第 2 个条件。

1. `data.withColumn("session",functions.when(data.col("refId").isNull(),true));
3.

WindowSpec w = Window.partitionBy("id, refid").orderBy(timestampDS.col("timestamp"));
functions.coalesce(timestampDS.col("timestamp").cast("long").$minus(functions.lag("timestamp", 1).over(w).cast("long")), functions.lit(0));

我的问题是如何满足第二个条件并同时实现所有 3 个转换。

Answer 1

我想说使用 Spark SQL 可以降低复杂性并轻松实现结果

df.createOrReplaceTempView("test")

spark.sql("select id,refId,timestamp,case when refId is null and id is not null then 'true' when id is not null and refId is not null and rank=1 then 'true' else 'false' end as session from  (select id,refId,timestamp, rank() OVER (PARTITION BY id,refId ORDER BY timestamp DESC) as rank from test) c").show()

输出如下所示:

+---+--------+----------+-------+
| id|   refId| timestamp|session|
+---+--------+----------+-------+
|  1|    null|1548944642|   true|
|  1|29950529|1548937685|   true|
|  2|27510720|1548944885|   true|
|  2|27510720|1548943617|  false|
+---+--------+----------+-------+