Spark Scala - Perform data aggregation on last or next n seconds time window



Often while performing statistical aggregations, we get scenario to perform aggregations on next n number of seconds from current row for each of the rows.




Spark Window api provides a nice rangeBetween functionality which facilitates performing above.




For Example:
// Sample data with timestamp

// Sample data with timestamp
val customers = sc.parallelize(List(("Alice", "2016-05-01 00:00:00", 10,4),
("Alice", "2016-05-01 00:00:01", 20,2),
("Alice", "2016-05-01 00:00:02", 30,4),
("Alice", "2016-05-01 00:00:02", 40,6),
("Alice", "2016-05-01 00:00:03", 50,1),
("Alice", "2016-05-01 00:00:03", 60,4),
("Alice", "2016-05-01 00:00:04", 70,2),
("Alice", "2016-05-01 00:00:05", 80,4),
("Bob", "2016-05-01 00:00:03", 25,6),
("Bob", "2016-05-01 00:00:04", 29,7),
("Bob", "2016-05-01 00:00:05", 27,10))).toDF("name", "time", "amountSpent","NumItems")


Spark Console output;scala> customers.show
+-----+-------------------+-----------+--------+
| name| time|amountSpent|NumItems|
+-----+-------------------+-----------+--------+
|Alice|2016-05-01 00:00:00| 10| 4|
|Alice|2016-05-01 00:00:01| 20| 2|
|Alice|2016-05-01 00:00:02| 30| 4|
|Alice|2016-05-01 00:00:02| 40| 6|
|Alice|2016-05-01 00:00:03| 50| 1|
|Alice|2016-05-01 00:00:03| 60| 4|
|Alice|2016-05-01 00:00:04| 70| 2|
|Alice|2016-05-01 00:00:05| 80| 4|
| Bob|2016-05-01 00:00:03| 25| 6|
| Bob|2016-05-01 00:00:04| 29| 7|
| Bob|2016-05-01 00:00:05| 27| 10|
+-----+-------------------+-----------+--------+

// Convert data to epoch, because Window orderBy column need to be a numeric type
val tdf = customers.withColumn("epoch", unix_timestamp($"time"))


// Create a Window partition from current row to next five seconds of data (inclusive)
val wSpec1 = Window.partitionBy("name").orderBy("epoch").rangeBetween(Window.currentRow, 5)


// Get the list of amountSpent by customers from now onwards till next 5 seconds
tdf.withColumn("sumSpent",collect_list(customers("amountSpent")).over(wSpec1)).show()
whose output:
scala> tdf.withColumn("sumSpent",collect_list(customers("amountSpent")).over(wSpec1)).show(false)
+-----+-------------------+-----------+--------+----------+--------------------------------+
|name |time |amountSpent|NumItems|epoch |sumSpent |
+-----+-------------------+-----------+--------+----------+--------------------------------+
|Bob |2016-05-01 00:00:03|25 |6 |1462024803|[25, 29, 27] |
|Bob |2016-05-01 00:00:04|29 |7 |1462024804|[29, 27] |
|Bob |2016-05-01 00:00:05|27 |10 |1462024805|[27] |
|Alice|2016-05-01 00:00:00|10 |4 |1462024800|[10, 20, 30, 40, 50, 60, 70, 80]|
|Alice|2016-05-01 00:00:01|20 |2 |1462024801|[20, 30, 40, 50, 60, 70, 80] |
|Alice|2016-05-01 00:00:02|30 |4 |1462024802|[30, 40, 50, 60, 70, 80] |
|Alice|2016-05-01 00:00:02|40 |6 |1462024802|[30, 40, 50, 60, 70, 80] |
|Alice|2016-05-01 00:00:03|50 |1 |1462024803|[50, 60, 70, 80] |
|Alice|2016-05-01 00:00:03|60 |4 |1462024803|[50, 60, 70, 80] |
|Alice|2016-05-01 00:00:04|70 |2 |1462024804|[70, 80] |
|Alice|2016-05-01 00:00:05|80 |4 |1462024805|[80] |
+-----+-------------------+-----------+--------+----------+--------------------------------+
// Same as above, but sum of amountSpent
tdf.withColumn("sumSpent",sum(customers("amountSpent")).over(wSpec1)).show()


Comments

Post a Comment

Popular posts from this blog

Spark Cluster Mode - Too many open files

When running complex Spark Jobs in Cluster mode (Yarn-Client or Yarn-Cluster) mode, it is quite possible the default ulimit (number of open files) of 1024 is not sufficient and hence gets error "Too many open files". One way to address this is issue is by increasing the ulimit size. Steps (as root user): 1) vim /etc/sysctl.conf, append the file with:     fs.file-max = 65536 2) vim /etc/security/limits.conf, append the file with: * soft nproc 65535 * hard nproc 65535 * soft nofile 65535 * hard nofile 65535 3) ulimit -n 65535
Read more

Binary Data to Float using Spark SQL

Often when there is massive data communication among the enterprise applications, the data is converted to binary datatype, for reasons like: Faster I/O, Smaller Size, any type of data can be represented, etc. This brings in need to convert binary back to known/readable data type for further analysis/processing etc. For example, here floating point data is represented in Decimal Array[Byte] then stored in Binary format. Ex: [62, -10, 0, 0] Whole floating point equivalent is: 0.480 To achieve above conversion using Apache Spark Sql DataFrames: Voila!
Read more

HDFS filenames without rest of the file details

Get the list of file names and absolute path alone from HDFS: General hdfs list output would be: $ hdfs dfs -ls -rw-r--r--   3 foo bar    6346268 2016-12-28 02:52 /user/foo/data/file007.csv -rw-r--r--   3 foo bar    4397850 2016-12-28 02:52 /user/foo/data/file014.csv -rw-r--r--   3 foo bar   13297361 2016-12-28 02:52 /user/foo/data/file020.csv -rw-r--r--   3 foo bar   10400852 2016-12-28 02:53 /user/foo/data/file118.csv -rw-r--r--   3 foo bar   10184639 2016-12-28 02:52 /user/foo/data/file205.csv -rw-r--r--   3 foo bar    5542293 2016-12-28 02:53 /user/foo/data/file214.csv -rw-r--r--   3 foo bar    6085128 2016-12-28 02:53 /user/foo/data/file307.csv But we would need get just the absolute hdfs file paths especially in shell scripts to perform CRUD operations, like: /user/foo/data/file007.csv /user/foo/data/file014.csv /user/foo/data/file020.csv /user/foo/data/file118.csv /user/foo/data/file205.csv /user/foo/data/file214.csv One way to achieve this is using awk, like: $
Read more