CODEWITHSUNDEEP
apache-sparkpysparktime-seriesapache-spark-sql
serdar
serdar

Reputation: 560

Spark time series query with two date columns

I need to do some calculations based on historical data in Spark, but my case is a little different than examples that float all over the Internet. I have a dataset with 3 columns: enter_date, exit_date, client_id. I need to calculate online client counts between hourly intervals.

For example consider following data:

enter_date             | exit_date               | client_id
2017-03-01 12:30:00    | 2017-03-01 13:30:00     | 1
2017-03-01 12:45:00    | 2017-03-01 14:10:00     | 2
2017-03-01 13:00:00    | 2017-03-01 15:20:00     | 3

I must get following as result:

time_interval          | count
2017-03-01 12:00:00    | 2
2017-03-01 13:00:00    | 3
2017-03-01 14:00:00    | 2
2017-03-01 15:00:00    | 1

As you can see, calculation must be performed based on not only enter_date, but both enter_date and exit_date columns.

So, there are mainly 2 questions:

  1. Is spark able to do this type of calculations?
  2. If yes, how?

Upvotes: 2

Views: 339

Answers (2)

Japu_D_Cret
Japu_D_Cret

Reputation: 638

you can also do it with the Spark SQL, but there you have to use another Dataset which contains the intervals. I used a separate CSV File, but in theory you can add it however you want.

my setup

Apache Spark in Java

  • spark-core_2.10
  • spark-sql_2.10

Needed Files:

timeinterval.csv

time_interval
01.03.2017 11:00:00
01.03.2017 12:00:00
01.03.2017 13:00:00
01.03.2017 14:00:00
01.03.2017 15:00:00
01.03.2017 16:00:00

test.csv

enter_date             | exit_date               | client_id
2017-03-01 12:30:00    | 2017-03-01 13:30:00     | 1
2017-03-01 12:45:00    | 2017-03-01 14:10:00     | 2
2017-03-01 13:00:00    | 2017-03-01 15:20:00     | 3

how i did it

I did this in Java, but since I use SQL the conversion should be pretty simple

Dataset<Row> rowsTest = spark.read()
  .option("header", "true")
  .option("delimiter", ";")
  .option("quoteMode", "NONE")
  .csv("C:/Temp/stackoverflow/test.csv");

Dataset<Row> rowsTimeInterval = spark.read()
  .option("header", "true")
  .option("delimiter", ";")
  .option("quoteMode", "NONE")
  .csv("C:/Temp/stackoverflow/timeinterval.csv");

rowsTest.createOrReplaceTempView("test");
rowsTimeInterval.createOrReplaceTempView("timeinterval");

String sql = "SELECT timeinterval.time_interval,(" +
              "SELECT COUNT(test.client_id) FROM timeinterval AS sub" +
                " INNER JOIN test ON " +
                  " ((unix_timestamp(sub.time_interval,\"dd.MM.yyyy HH:mm:SS\") + 60*60) > unix_timestamp(test.enter_date,\"dd.MM.yyyy HH:mm:SS\"))" +
                  " AND" +
                  " (sub.time_interval < test.exit_date)" +
                " WHERE timeinterval.time_interval = sub.time_interval" +
              ") AS RowCount" +
              " FROM timeinterval";
Dataset<Row> result = spark.sql(sql);

result.show();

here the raw SQL statement

SELECT timeinterval.time_interval,(
    SELECT COUNT(test.client_id)
    FROM timeinterval AS sub
    INNER JOIN test ON
        (unix_timestamp(sub.time_interval,"dd.MM.yyyy HH:mm:SS") + 60*60) > unix_timestamp(test.enter_date,"dd.MM.yyyy HH:mm:SS"))
        AND
        (sub.time_interval < test.exit_date)
    WHERE
        timeinterval.time_interval = sub.time_interval
) AS RowCount
FROM timeinterval

as I make use of the unix_timestamp function(see https://spark.apache.org/docs/1.6.2/api/java/org/apache/spark/sql/functions.html#unix_timestamp%28%29) you'll need a version equal to or higher than 1.5.0

the result

|      time_interval|RowCount|
+-------------------+--------+
|01.03.2017 11:00:00|       0|
|01.03.2017 12:00:00|       2|
|01.03.2017 13:00:00|       3|
|01.03.2017 14:00:00|       2|
|01.03.2017 15:00:00|       1|
|01.03.2017 16:00:00|       0|
+-------------------+--------+

Upvotes: 1

pasha701
pasha701

Reputation: 7207

On Scala can be implemented in this way, guess, Python is similar:

val clientList = List(
  Client("2017-03-01 12:30:00", "2017-03-01 13:30:00", 1),
  Client("2017-03-01 12:45:00", "2017-03-01 14:10:00", 2),
  Client("2017-03-01 13:00:00", "2017-03-01 15:20:00", 3)
)

val clientDF = sparkContext.parallelize(clientList).toDF
val timeFunctions = new TimeFunctions()

val result = clientDF.flatMap(
   // return list of times between "enter_date" and "exit_date"
  row => timeFunctions.getDiapason(row.getAs[String]("enter_date"), row.getAs[String]("exit_date"))
).map(time => (time, 1)).reduceByKey(_ + _).sortByKey(ascending = true)

result.foreach(println(_))

Result is folowing:

(2017-03-01 12:00:00,2)
(2017-03-01 13:00:00,3)
(2017-03-01 14:00:00,2)
(2017-03-01 15:00:00,1)

TimeFunctions can be implemented like:

  val formatter = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss")
  def getDiapason(from: String, to: String): Seq[String] = {
      var fromDate = LocalDateTime.parse(from,formatter).withSecond(0).withMinute(0)
      val result = ArrayBuffer(formatter.format(fromDate))

      val toDate = LocalDateTime.parse(to, formatter).withSecond(0).withMinute(0)
      while (toDate.compareTo(fromDate) > 0) {
        fromDate = fromDate.plusHours(1)
        result += formatter.format(fromDate)
      }
      result
    }

Upvotes: 2

Related Questions