Spark: получить метку времени для окончания действия / записи

Question

У меня есть конвейер Spark с 3 DataFrame.write, и я хочу вывести временную метку для каждой записи. Я пытался так:

val timeFormat = new SimpleDataFormat("HH:mm:ss")
val calendar = Calendar.getInstance()

df.groupby(...)
  .agg(...)
  .write
  .csv(...)

println(timeFormat.format(calendar.getTime))

Но то, что я вижу в выводе, выглядит так:

2018-07-04 12:31:55 INFO  DAGScheduler:54 - Job 5 finished ...
12:25:56

Мне нужен настоящий конец работы, поэтому 12:31:55 здесь. Как я могу получить правильное время?

Почему это происходит? Я предполагаю, что это как-то связано со скриптом драйвера, который просто порождает процессы-исполнители, но продолжает работать, фактически не дожидаясь окончания работы, прежде чем выполнять операторы print. Это правильно?

Answer 1

Как правильно заметил Манодж Кумар, получите время после выполнения, просто добавьте, используйте функцию, чтобы получить время окончания, чтобы вы могли использовать повторно.

scala> import java.text.SimpleDateFormat
import java.text.SimpleDateFormat

scala>

scala> def getEndTime[T](fn: => T): (T, String) = {
     |     val result = fn
     |     (result, new SimpleDateFormat("yyyy/MM/dd HH:mm:ss").format(System.currentTimeMillis()))
     | }
getEndTime: [T](fn: => T)(T, String)

scala>

scala> def add(a : Int, b : Int) : Int = {
     |     a + b
     | }
add: (a: Int, b: Int)Int

scala>

scala> val (result, endTime) = getEndTime { add(10, 5) }
result: Int = 15
endTime: String = 2018/07/04 07:18:58

scala>

scala> val (result, endTime) = getEndTime { 10 + 5 }
result: Int = 15
endTime: String = 2018/07/04 07:19:00

Чтобы вы могли написать что-то вроде:

val (_, endTime) = getEndTime { df.groupby(...)
                    .agg(...)
                    .write
                    .csv(...)}

Answer 2

Вы должны занять некоторое время сразу после функции write, как показано ниже

    import java.text.SimpleDateFormat

    df.groupby(...)
      .agg(...)
      .write
      .csv(...)

    val currentTime=new SimpleDateFormat("yyyyMMdd HH:mm:ss").format(System.currentTimeMillis())