Записи отсутствуют после создания таблицы из временной таблицы спарка в Spark2 - PullRequest
Новая
       37

Записи отсутствуют после создания таблицы из временной таблицы спарка в Spark2

0 голосов
/ 17 декабря 2018

Я создал последовательность данных снизу.

val df = sc.parallelize(Seq((100,23,9.50),
(100,23,9.51),
(100,24,9.52),
(100,25,9.54),
(100,23,9.55),
(101,21,8.51),
(101,23,8.52),
(101,24,8.55),
(101,20,8.56))).toDF("id", "temp","time")

Я хотел обновить DF, добавив еще несколько строк, где данные отсутствуют на данный момент.Поэтому я итерировал DF из mapPartitions, чтобы добавить новые строки.

import org.apache.spark.sql.functions._
import org.apache.spark.sql.{DataFrame, Row, Column}

@transient val w = org.apache.spark.sql.expressions.Window.partitionBy("id").orderBy("time")

val leadDf = df.withColumn("time_diff", ((lead("time", 1).over(w) - df("time")).cast("Float")*100).cast("int"))

Здесь выполняется итерация данных: 

val result =   leadDf.rdd.mapPartitions(itr =>
  new Iterator[Row] {
    var prevRow = null: Row
    var prevDone = true
    var firstRow = true
    var outputRow: Row = null: Row
    var counter  = 0
    var currRecord = null :Row
    var currRow: Row = if (itr.hasNext) {currRecord = itr.next;  currRecord } else null
    prevRow = currRow
    override def hasNext: Boolean = {
      if (!prevDone) {
        prevRow = incrementValue(prevRow,2)
        outputRow = prevRow
        counter = counter -1
        if(counter == 0) {
          prevDone = true
        }
        true
      } else if (itr.hasNext) {
        prevRow = currRow
        if(counter == 0 && prevRow.getAs[Int](3) != 1 && !isNullValue(prevRow,3 )){
          outputRow = prevRow
          counter = prevRow.getAs[Int](3) - 1
          prevDone = false
        }else if(counter > 0) {
          counter = counter -1
          prevDone = false
        }
        else {
          outputRow = currRow
        }
        //if(counter == 0){
        currRow = itr.next
        true
      } else if (currRow != null) {
        outputRow = currRow
        currRow =null
        true
      } else {
        false
      }
    }
    override def next(): Row = outputRow
  })
  val newDf = spark.createDataFrame(result,leadDf.schema)

После этого я вижу 12 записей в кадре данных.Но получил 10 записей из физической таблицы, созданной временной таблицей, созданной из фрейма данных "newDf". 

newDf.registerTempTable("test")
spark.sql("create table newtest as select * from test")

scala> newDf.count
res14: Long = 12

scala> spark.sql("select * from newtest").count
res15: Long = 10

Тот же код отлично работает в Spark 1.6, и итоговое число таблиц соответствует количеству записей в фрейме данных.

Может кто-нибудь объяснить, почему это происходит?и любое решение или обходной путь для решения проблемы

1 Ответ

0 голосов
/ 17 декабря 2018

Я нашел решение или обходной путь, который вызывает метод репарации на вновь созданном фрейме данных из RDD [Row].

val newDf = spark.createDataFrame(result,leadDf.schema).repartition(result.getNumPartitions)
...