Разбить одну строку на несколько строк

Question

Я хочу преобразовать одну строку из кадра данных в несколько строк.Если часы одинаковы, то строки не будут разбиты, но если часы разные, то строки будут разбиты на несколько строк по разнице между часами. У меня все хорошо с решением, использующим функцию dataframe или запрос улья.

Input Table или Dataframe

Таблица ожидаемых результатов или структура данных

Пожалуйста, помогите мне обойти ожидаемый результат.

Answer 1

Вы можете реализовать свою собственную логику в операции map и использовать flatMap для достижения этой цели.

Вот грубый способ, которым я реализовал решение, вы можете импровизироватьэто по необходимости.

import java.time.format.DateTimeFormatter
import java.time.temporal.ChronoUnit
import java.time.{Duration, LocalDateTime}

import org.apache.spark.sql.Row

import scala.collection.mutable.ArrayBuffer

import sparkSession.sqlContext.implicits._

val df = Seq(("john", "2/9/2018", "2/9/2018 5:02", "2/9/2018 5:12"),
    ("smit", "3/9/2018", "3/9/2018 6:12", "3/9/2018 8:52"),
    ("rick", "4/9/2018", "4/9/2018 23:02", "5/9/2018 2:12")
  ).toDF("UserName", "Date", "start_time", "end_time")

val rdd = df.rdd.map(row => {
  val result = new ArrayBuffer[Row]()
  val formatter1 = DateTimeFormatter.ofPattern("d/M/yyyy H:m")
  val formatter2 = DateTimeFormatter.ofPattern("d/M/yyyy H:mm")

  val d1 = LocalDateTime.parse(row.getAs[String]("start_time"), formatter1)
  val d2 = LocalDateTime.parse(row.getAs[String]("end_time"), formatter1)

  if (d1.getHour == d2.getHour) result += row
  else {
    val hoursDiff = Duration.between(d1, d2).toHours.toInt

    result += Row.fromSeq(Seq(
      row.getAs[String]("UserName"),
      row.getAs[String]("Date"),
      row.getAs[String]("start_time"),
      d1.plus(1, ChronoUnit.HOURS).withMinute(0).format(formatter2)))

    for (index <- 1 until hoursDiff) {
      result += Row.fromSeq(Seq(
        row.getAs[String]("UserName"),
        row.getAs[String]("Date"),
        d1.plus(index, ChronoUnit.HOURS).withMinute(0).format(formatter1),
        d1.plus(1 + index, ChronoUnit.HOURS).withMinute(0).format(formatter2)))
    }

    result += Row.fromSeq(Seq(
      row.getAs[String]("UserName"),
      row.getAs[String]("Date"),
      d2.withMinute(0).format(formatter2),
      row.getAs[String]("end_time")))
  }
  result
}).flatMap(_.toIterator)

rdd.collect.foreach(println)

и, наконец, ваш результат выглядит следующим образом:

[john,2/9/2018,2/9/2018 5:02,2/9/2018 5:12]
[smit,3/9/2018,3/9/2018 6:12,3/9/2018 7:00]
[smit,3/9/2018,3/9/2018 7:0,3/9/2018 8:00]
[smit,3/9/2018,3/9/2018 8:00,3/9/2018 8:52]
[rick,4/9/2018,4/9/2018 23:02,5/9/2018 0:00]
[rick,4/9/2018,5/9/2018 0:0,5/9/2018 1:00]
[rick,4/9/2018,5/9/2018 1:0,5/9/2018 2:00]
[rick,4/9/2018,5/9/2018 2:00,5/9/2018 2:12]

Answer 2

Самым простым решением для такой простой схемы является использование Dataset.flatMap после определения классов дел для схемы ввода и вывода.

Простое решение UDF вернет последовательность, а затем вы можете использовать functions.explode. Гораздо менее чистый и эффективный, чем при использовании flatMap.

И последнее, но не менее важное: вы могли бы создать свой собственный UDF для генерации таблиц, но это было бы излишним излишним для этой проблемы.