Operating on Multiple Rows in Apache Spark SQL

Question

I am trying to create a function in Apache Spark SQL that operates on multiple rows of data but have not been able to find a way of doing this directly in Spark - in Java.

My current solution has been to extract the data out of Spark DataFrames and into Java Lists for processing before returning back to a Spark DataFrame. This is not ideal performance-wise.

The best option seems to be Window functions, but unfortunately these require Hive context, which I don't have access to. The explode() function seems another option, but again, this is Scala-specific and I was unable to get it working in Java.

Perhaps this could be done by converting the DataFrame back into an RDD?

If anyone has any tips or suggestions as to how this may be done for Apache Spark SQL in Java, that would be greatly appreciated. Thank you.

Update: Example provided:

+----------+-----------+------------+
|   Item   | Timestamp | Difference |
+----------+-----------+------------+
|     A    |   11:00   |    02:00   |
|     A    |   13:00   |      -     |
+----------+-----------+------------+
|     B    |   09:00   |      -     |
+----------+-----------+------------+
|     C    |   15:15   |    00:20   |
|     C    |   15:35   |    01:30   |
|     C    |   17:05   |      -     |
+----------+-----------+------------+

So in the example, I am trying to operate on pairs of rows, grouped by Item, to calculate the time difference between each item row.

Such a task is possible with LAG() and LEAD() functions in SQL, but these require Hive in Spark.

Can you provide an example of how you'd like it to work? Do you want a function to apply over the entire rowset or just a part? My first thought is depending on the complexity it sounds like a GROUPBY with a UDF. — Chet
Apologies for not being clear - I am trying to operate over several rows and not the entire row set. I have provided an example above as requested. — ab853

Chet Chet · Accepted Answer · 2015-12-08T00:44:04

As of Spark 1.5 you can now define UDAFs, or User Defined Aggregate Functions to let you perform custom aggregations over groups of input data. I think this is probably the closest thing I've seen to what you're looking for.

Generally you'll need to create a class that extends UserDefinedAggregateFunction and implements the required methods which involve initialization, merging, and aggregating.

Once you've created that you can just instanciate it, register it, and then use it within your SQL.

val myAggregation = new MyAggregation 
sqlContext.udf.register("MY_AGG", myAggregation)

https://databricks.com/blog/2015/09/16/spark-1-5-dataframe-api-highlights-datetimestring-handling-time-intervals-and-udafs.html

Operating on Multiple Rows in Apache Spark SQL

1 Answers