spark檢測并提取列值中的模式-有解無憂

我有一個這樣的 df

    import spark.implicits._
    import org.apache.spark.sql.functions._
    
    val latenies = Seq(
        ("start","304875","2021-10-25 21:26:23.486027"),
        ("start","304875","2021-10-25 21:26:23.486670"),
        ("end","304875","2021-10-25 21:26:23.487590"),
        ("start","304875","2021-10-25 21:26:23.509683"),
        ("end","304875","2021-10-25 21:26:23.509689"),
        ("end","304875","2021-10-25 21:26:23.510154"),
        ("start","201345","2021-10-25 21:26:23.510156"),
        ("end","201345","2021-10-25 21:26:23.510159"),
        ("start","201345","2021-10-25 21:26:23.510333"),
        ("start","201345","2021-10-25 21:26:23.510335"),
        ("end","201345","2021-10-25 21:26:23.513177"),
        ("start","201345","2021-10-25 21:26:23.513187")
      )
    val latenies_df = latenies.toDF("Msg_name","Id_num","TimeStamp")
                            .withColumn("TimeStamp", to_timestamp(col("TimeStamp")))
    latenies_df.show(false)

它看起來像這樣：

 -------- ------ -------------------------- 
|Msg_name|Id_num|TimeStamp                 |
 -------- ------ -------------------------- 
|start   |304875|2021-10-25 21:26:23.486027|
|start   |304875|2021-10-25 21:26:23.48667 |
|end     |304875|2021-10-25 21:26:23.48759 |
|start   |304875|2021-10-25 21:26:23.509683|
|end     |304875|2021-10-25 21:26:23.509689|
|end     |304875|2021-10-25 21:26:23.510154|
|start   |201345|2021-10-25 21:26:23.510156|
|end     |201345|2021-10-25 21:26:23.510159|
|start   |201345|2021-10-25 21:26:23.510333|
|start   |201345|2021-10-25 21:26:23.510335|
|end     |201345|2021-10-25 21:26:23.513177|
|start   |201345|2021-10-25 21:26:23.513187|
 -------- ------ --------------------------

問題：我想在列中提取某個模式，Msg_name該模式始終為 when 的start后續值為endwhen partitioned byId和 order by time。 Msg可以有多個開始一個接一個或結束。我只想要start-end兩者之間。

使用這種模式，我想做一個 df ：

|patter_name|Timestamp_start           |Timestamp_end             |Id_num  |
|   pattern1|2021-10-25 21:26:23.486670|2021-10-25 21:26:23.487590|304875  |
|   pattern1|2021-10-25 21:26:23.509683|2021-10-25 21:26:23.509689|304875  |
|   pattern1|2021-10-25 21:26:23.510156|2021-10-25 21:26:23.510159|201345  |
|   pattern1|2021-10-25 21:26:23.510335|2021-10-25 21:26:23.513177|201345  |

我所做的是移動框架，由于Msg_name專欄的性質，這不會給我正確的答案。

    val window = org.apache.spark.sql.expressions.Window.partitionBy("Id_num").orderBy("TimeStamp")
    val df_only_pattern = latenies_df.withColumn("TimeStamp_start", when($"Msg_name" !== lag($"Msg_name", 1).over(window), lag("TimeStamp", 1).over(window)).otherwise(lit(null)))
                                    .withColumn("latency_time", when($"TimeStamp_start".isNotNull, round((col("TimeStamp").cast("double")-col("TimeStamp_start").cast("double")) * 1e3, 2)).otherwise(lit(null)))
                                    .withColumnRenamed("TimeStamp", "TimeStamp_end")
                                    .withColumn("patter_name", lit("pattern1"))
                                    .na.drop()
    df_only_pattern.orderBy("TimeStamp_start").show(false)

這給出了什么：

 -------- ------ -------------------------- -------------------------- ------------ ----------- 
|Msg_name|Id_num|TimeStamp_end             |TimeStamp_start           |latency_time|patter_name|
 -------- ------ -------------------------- -------------------------- ------------ ----------- 
|end     |304875|2021-10-25 21:26:23.48759 |2021-10-25 21:26:23.48667 |0.92        |pattern1   |
|start   |304875|2021-10-25 21:26:23.509683|2021-10-25 21:26:23.48759 |22.09       |pattern1   |
|end     |304875|2021-10-25 21:26:23.509689|2021-10-25 21:26:23.509683|0.01        |pattern1   |
|end     |201345|2021-10-25 21:26:23.510159|2021-10-25 21:26:23.510156|0.0         |pattern1   |
|start   |201345|2021-10-25 21:26:23.510333|2021-10-25 21:26:23.510159|0.17        |pattern1   |
|end     |201345|2021-10-25 21:26:23.513177|2021-10-25 21:26:23.510335|2.84        |pattern1   |
|start   |201345|2021-10-25 21:26:23.513187|2021-10-25 21:26:23.513177|0.01        |pattern1   |
 -------- ------ -------------------------- -------------------------- ------------ -----------

我可以df使用帶有 groupby 的 python pandas 并在組內回圈來實作想要的，這在 spark 中似乎是不可能的。

uj5u.com熱心網友回復：

可以獲取訊息“結束”，其前一行帶有“開始”：

latenies_df
  .withColumn("TimeStamp_start",
    when(lag($"Msg_name", 1).over(window) === lit("start"), lag($"TimeStamp", 1).over(window))
      .otherwise(lit(null).cast(TimestampType))
  )
  .where($"Msg_name" === lit("end"))
  .where($"TimeStamp_start".isNotNull)

  .select(
    lit("pattern1").alias("patter_name"),
    $"TimeStamp_start",
    $"TimeStamp".alias("Timestamp_end"),
    $"Id_num"
  )

結果：

 ----------- -------------------------- -------------------------- ------ 
|patter_name|TimeStamp_start           |Timestamp_end             |Id_num|
 ----------- -------------------------- -------------------------- ------ 
|pattern1   |2021-10-25 21:26:23.48667 |2021-10-25 21:26:23.48759 |304875|
|pattern1   |2021-10-25 21:26:23.509683|2021-10-25 21:26:23.509689|304875|
|pattern1   |2021-10-25 21:26:23.510156|2021-10-25 21:26:23.510159|201345|
|pattern1   |2021-10-25 21:26:23.510335|2021-10-25 21:26:23.513177|201345|
 ----------- -------------------------- -------------------------- ------

轉載請註明出處，本文鏈接：https://www.uj5u.com/caozuo/339694.html

標籤：斯卡拉阿帕奇火花 apache-spark-sql

上一篇：Scala獲得結果的平均值

下一篇：如何使用scala獲取串列中重復對的索引