how to use spark to count continuous identical event?

user_1235372 0
Viewed 100

I have a user action DataFrame, like

+-------+-----+------+
|user_id|stamp| event|
+-------+-----+------+
|   usr1|    1|gather|
|   usr1|    2|  kill|
|   usr1|    3|  kill|
|   usr1|    4|  jump|
|   usr1|    5|  kill|
|   usr1|    6|  kill|
|   usr2|    1|  jump|
|   usr2|    2|  jump|
|   usr2|    3|gather|
|   usr2|    4|  jump|
+-------+-----+------+

I want to count continuous identical event, get a DataFrame like

+----+------+-----------+
|user| event|event_count|
+----+------+-----------+
|usr1|gather|          1|
|usr1|  kill|          2|
|usr1|  jump|          1|
|usr1|  kill|          2|
|usr2|  jump|          2|
|usr2|gather|          1|
|usr2|  jump|          1|
+----+------+-----------+

I want to count continuous identical event of a user, not count event of a user, how to do this by Spark?

What's more, event sequence of one user is very very large(max 0.4 million), groupBy method maybe OOM.

tks~

apache-spark
1 Answers

One way of doing it is via RDD by :

  1. Repartition and group data of a single user
  2. Now iterate over it and apply your Own logic to get your required rows
  3. Flatmap it to flatten your rdd
  4. convert back to DF
import org.apache.spark.sql.Row;
import scala.collection.mutable.Seq;
import org.apache.spark.sql.types.{IntegerType, StringType, StructField, StructType}

// assuming your initial df is called "userDf" with ["user_id", "stamp","event"]

val pairRdd = userDf.rdd.map(row => (row(0).toString,row))
val groupedRdd = pairRdd.groupByKey();

// your user related data are brought together in an partition 
// now apply your complex logic for each parititon 

val rddOfInteratorOfRow = groupedRdd.map{
      (it) => {

        var sortedlist = it._2.toList.sortBy( x => x.get(1).asInstanceOf[Int])
        var mySeq : Seq[Row] = Seq.empty

        var count=1;
        var prevEvent = sortedlist.apply(0).apply(2).toString;
        sortedlist = sortedlist.drop(1);

        for( row <- sortedlist){

          if(row(2).equals(prevEvent)){
            count=count+1;
          }
          else{
            mySeq = mySeq:+Row(it._1,prevEvent,count);
            count = 1;
            prevEvent = row(2).toString;
          }
        }
        mySeq = mySeq:+Row(it._1,prevEvent,count);

        mySeq.toIterator
      }
    }

val flattendRdd = rddOfInteratorOfRow.flatMap(x =>x)

val schema = new StructType()
      .add(StructField("user_id", StringType, true))
      .add(StructField("event", StringType, true))
      .add(StructField("count", IntegerType, true))

val finaldf = spark.createDataFrame(flattendRdd,schema)
/*
scala> finaldf.show
+-------+------+-----+
|user_id| event|count|
+-------+------+-----+
|   usr1|gather|    1|
|   usr1|  kill|    2|
|   usr1|  jump|    1|
|   usr1|  kill|    2|
|   usr2|  jump|    2|
|   usr2|gather|    1|
|   usr2|  jump|    1|
+-------+------+-----+
*/

This might be the most efficient way, but I am unaware of any DF based shorter and efficient solution out of the box by spark.

user_70945200
Related