如何對不相關的內容進行分組

我有一個記錄用戶行為的配置單元表

像這樣

每天大約有 400GB 被添加到表中。

我想按時間戳 asc 排序，然后一個“視圖”位于另一個“視圖”之間的組中，如此表，前 3 行屬于同一組，然后減去時間戳，如 1650022614 - 1650022601 作為查看時間。

這該怎么做？

我嘗試滯后和領先功能，或者像這樣的scala

        val pairRDD: RDD[(Int, String)] = record.map(x => {
            if (StringUtil.isDateString(x.split("\\s ")(0))) {
                partition = partition   1
                (partition, x)
            } else {
                (partition, x)
            }
        })

或者像這樣的java

        LongAccumulator part = spark.sparkContext().longAccumulator("part");

        JavaPairRDD<Long, Row> pairRDD = spark.sql(sql).coalesce(1).javaRDD().mapToPair((PairFunction<Row, Long, Row>) row -> {
            if (row.getAs("event") == "pageview") {
                part.add(1L);
            }
        return new Tuple2<>(part.value(), row);
        });

但是當資料集非常大時，這段代碼就很愚蠢。

請救救我

uj5u.com熱心網友回復：

如果您使用資料框，則可以使用視窗來構建磁區，該視窗對更改磁區時值為 1 和不更改磁區時值為 0 的列求和。

您可以使用本答案sparkSession.createDataframe()中解釋的方法將 RDD 轉換為資料幀

回到你的問題。在您的情況下，每次列行為等于“視圖”時，您都會更改磁區。所以我們可以從這個條件開始：

import org.apache.spark.sql.functions.col

val df1 = df.withColumn("is_view", (col("behavior") === "view").cast("integer"))

您將獲得以下資料框：

 ------ -------- ---------- ---- ------- 
|userid|behavior|timestamp |url |is_view|
 ------ -------- ---------- ---- ------- 
|1     |view    |1650022601|url1|1      |
|1     |click   |1650022602|url2|0      |
|1     |click   |1650022614|url3|0      |
|1     |view    |1650022617|url4|1      |
|1     |click   |1650022622|url5|0      |
|1     |view    |1650022626|url7|1      |
|2     |view    |1650022628|url8|1      |
|2     |view    |1650022631|url9|1      |
 ------ -------- ---------- ---- ------- 

然后使用按時間戳排序的視窗對is_view列求和：

import org.apache.spark.sql.expressions.Window
import org.apache.spark.sql.functions.sum

val df2 = df1.withColumn("partition", sum("is_view").over(Window.partitionBy("userid").orderBy("timestamp")))

這將為您提供以下資料框：

 ------ -------- ---------- ---- ------- --------- 
|userid|behavior|timestamp |url |is_view|partition|
 ------ -------- ---------- ---- ------- --------- 
|1     |view    |1650022601|url1|1      |1        |
|1     |click   |1650022602|url2|0      |1        |
|1     |click   |1650022614|url3|0      |1        |
|1     |view    |1650022617|url4|1      |2        |
|1     |click   |1650022622|url5|0      |2        |
|1     |view    |1650022626|url7|1      |3        |
|2     |view    |1650022628|url8|1      |1        |
|2     |view    |1650022631|url9|1      |2        |
 ------ -------- ---------- ---- ------- --------- 

然后，您只需匯總每個用戶 ID 和磁區：

import org.apache.spark.sql.functions.{max, min}

val result = df2.groupBy("userid", "partition")
  .agg((max("timestamp") - min("timestamp")).as("duration"))

你會得到以下結果：

 ------ --------- -------- 
|userid|partition|duration|
 ------ --------- -------- 
|1     |1        |13      |
|1     |2        |5       |
|1     |3        |0       |
|2     |1        |0       |
|2     |2        |0       |
 ------ --------- -------- 

完整的scala代碼：

import org.apache.spark.sql.expressions.Window
import org.apache.spark.sql.functions.{col, max, min, sum}

val result = df
  .withColumn("is_view", (col("behavior") === "view").cast("integer"))
  .withColumn("partition", sum("is_view").over(Window.partitionBy("userid").orderBy("timestamp")))
  .groupBy("userid", "partition")
  .agg((max("timestamp") - min("timestamp")).as("duration"))

標籤：爪哇 斯卡拉 阿帕奇火花 蜂巢 大数据

上一篇：在pysparkDataFrame中創建日期范圍

下一篇：scala樹中的遞回方式（（密封特征）