Сериализация на RDD против DataFrame Spark - PullRequest
Новая
       18

Сериализация на RDD против DataFrame Spark

0 голосов
/ 09 марта 2019

EX1 . Это с RDD дает сериализацию, как мы ожидаем, с объектом или без него, при этом виновным является valnum, отлично: 

object Example {
 val r = 1 to 1000000 toList
 val rdd = sc.parallelize(r,3)
 val num = 1
 val rdd2 = rdd.map(_ + num)
 rdd2.collect  
}
Example

EX2 . Однако использование Dataframe аналогичным образом не позволяет. Почему это выглядит примерно так же? Что мне здесь не хватает? 

object Example {
import spark.implicits._
import org.apache.spark.sql.functions._

val n = 1 
val df = sc.parallelize(Seq(
    ("r1", 1, 1),
    ("r2", 6, 4),
    ("r3", 4, 1),
    ("r4", 1, 2)
    )).toDF("ID", "a", "b")
df.repartition(3).withColumn("plus1", $"b" + n).show(false)
}
Example

Причины, не совсем понятные для меня в отношении DF, могли бы ожидать аналогичного поведения. Похоже, DS обходят некоторые проблемы, но я вполне могу что-то упустить.

Работа на Databricks дает множество проблем с сериализацией, так что не думайте, что это влияет на вещи, пригодные для тестирования.

1 Ответ

2 голосов
/ 09 марта 2019

Причина проста и более фундаментальна, чем различие между RDD и Dataset:

  • Первый фрагмент кода оценивает функцию 

    _ + num

    поэтому его необходимо вычислить и оценить.

  • Второй фрагмент кода не. После 

    $"b" + n

    - это просто значение, поэтому вычисление замыканий и последующая сериализация не требуются.

Если это все еще не ясно, вы можете подумать об этом следующим образом:

  • Первый фрагмент кода сообщает Spark , как что-то сделать .
  • Последний фрагмент кода сообщает Spark , что делать . Фактический код, который выполняется, генерируется в другой области.

Если ваш Dataset код был ближе к его RDD аналогу, например: 

object Example {
  import spark.implicits._

  val num = 1
  spark.range(1000).map(_ + num).collect
}

или 

Example {
  import spark.implicits._
  import org.apache.spark.sql.functions._ 

  val num = 1
  val f = udf((x: Int) => x + num) 
  spark.range(1000).select(f($"id")).collect
}

произойдет сбой с исключением сериализации, как и в версии RDD.

...