Как лучше всего прочитать файл и преобразовать его в набор данных Spark sqlContext - PullRequest
Новая
       4

Как лучше всего прочитать файл и преобразовать его в набор данных Spark sqlContext

1 голос
/ 08 мая 2020

Я хочу прочитать файл, разделенный табуляцией, без заголовка (примеры строк, как показано ниже) 

196 242 3   881250949
186 302 3   891717742

У меня есть 2 решения для чтения файла и преобразования его в набор данных. Кто-нибудь может сказать мне, какое решение лучше и почему?

Solution1 

final case class Movie(movieID: Int)

import spark.implicits._

val moviesDS1 = spark.sparkContext.textFile("file path")
                .map(x => Movie(x.split("\t")(1).toInt))
                .toDS
                .select("movieID")

Solution 2 

final case class Movie(movieID: Int, Somenum1:Int, Somenum2: Int, Somenum3:Int)

import spark.implicits._
var movieSchema = Encoders.product[Movie].schema

val moviesDS2 = spark.read.options(Map("delimiter" -> "\t"))
                .schema(movieSchema)
                .csv("file path")
                .select("movieID")

Ответы [ 2 ]

1 голос
/ 16 мая 2020

Хорошо, давайте проведем небольшой тест, чтобы узнать, что лучше, 

 final case class Movie(movieID: Int)
  exec {

    import spark.implicits._

    val moviesDS1 = spark.sparkContext.textFile("mydata.csv/Movies.csv").toDS()
      .map { x => {
        Movie(x.split("\t")(0).toInt)

      }
      }
      .select("movieID").show(false)
   // moviesDS1.show(false)
  }
  final case class Movie1(movieID: Int, Somenum1: Int, Somenum2: Int, Somenum3: Int)
  exec {
    var movieSchema = Encoders.product[Movie1].schema

    val moviesDS2 = spark.read.options(Map("delimiter" -> "\t"))
      .schema(movieSchema)
      .csv("mydata.csv/Movies.csv")
      .select("movieID")
    moviesDS2.show(false)
  }

где exe c metthod будет измерять время в наносекундах ... 

  /**
    *
    * @param f
    * @tparam T
    * @return
    */
  def exec[T](f: => T) = {

    val starttime = System.nanoTime()
    println("t = " + f)
    val endtime = System.nanoTime()
    val elapsedTime = (endtime - starttime)
    //    import java.util.concurrent.TimeUnit
    //    val convertToSeconds = TimeUnit.MINUTES.convert(elapsedTime, TimeUnit.NANOSECONDS)
    println("time Elapsed " + elapsedTime)
  }

Результат: 

+-------+
|movieID|
+-------+
|196    |
|186    |
+-------+

t = ()
time Elapsed 5053611258
+-------+
|movieID|
+-------+
|196    |
|186    |
+-------+

t = ()
time Elapsed 573163989

Заключение:

По цифрам второй подход быстрее / оптимизирован (с 573163989 nano se c <5053611258 nano se c), чем первый подход. </p>

Решение 1 Мы должны позаботиться о синтаксическом анализе и сопоставлении с соответствующими классами, где, как и в решении 2, он исключает низкоуровневый синтаксический анализ и сопоставление данных с классами случаев.

Таким образом, решение 2 является лучшим вариантом.

Итак, ответ @QuickSilvers подходит для этого тестового примера.

1 голос
/ 08 мая 2020
  • Решение 2 всегда будет как минимум в 5 раз быстрее, чем Решение 1.
  • Решение 2 также обеспечивает неявную проверку ваших входных данных и помечает все значения столбцов нулевыми, если есть один несоответствие схемы.
  • Решение 2 также использует расширенные API-интерфейсы, которые предоставляют вашим фреймам данных похожие решения. Решение сначала загружает данные как RDD, а затем преобразует их в DataSet.
...