Преобразование списка scala в DataFrame или DataSet

Я новичок в Scala. Я пытаюсь преобразовать список scala (который содержит результаты некоторых вычисленных данных в исходном DataFrame) в Dataframe или набор данных. Я не нахожу никакого прямого метода сделать это. Тем не менее, я попробовал следующий процесс, чтобы преобразовать свой список в DataSet, но, похоже, он не работает. Я предоставляю 3 ситуации ниже.

Может кто-нибудь дать мне луч надежды, как сделать это преобразование? Спасибо.

import org.apache.spark.sql.{DataFrame, Row, SQLContext, DataFrameReader}
import java.sql.{Connection, DriverManager, ResultSet, Timestamp}
import scala.collection._

case class TestPerson(name: String, age: Long, salary: Double)
var tom = new TestPerson("Tom Hanks",37,35.5)
var sam = new TestPerson("Sam Smith",40,40.5)

val PersonList = mutable.MutableList[TestPerson]()

//Adding data in list
PersonList += tom
PersonList += sam

//Situation 1: Trying to create dataset from List of objects:- Result:Error
//Throwing error
var personDS = Seq(PersonList).toDS()
/*
ERROR:
error: Unable to find encoder for type stored in a Dataset.  Primitive types
   (Int, String, etc) and Product types (case classes) are supported by     
importing sqlContext.implicits._  Support for serializing other types will  
be added in future releases.
     var personDS = Seq(PersonList).toDS()

*/
//Situation 2: Trying to add data 1-by-1 :- Result: not working as desired.    
the last record overwriting any existing data in the DS
var personDS = Seq(tom).toDS()
personDS = Seq(sam).toDS()

personDS += sam //not working. throwing error


//Situation 3: Working. However, I am having consolidated data in the list    
which I want to convert to DS; if I loop the results of the list in comma  
separated values and then pass that here, it will work but will create an  
extra loop in the code, which I want to avoid.
var personDS = Seq(tom,sam).toDS()
scala> personDS.show()
+---------+---+------+
|     name|age|salary|
+---------+---+------+
|Tom Hanks| 37|  35.5|
|Sam Smith| 40|  40.5|
+---------+---+------+

apache-spark apache-spark-sql scala apache-spark-dataset apache-spark-encoders
person Leo    schedule 08.09.2016    source источник
comment
Какая у вас версия spark и scala?   -  person Ajeet Shah    schedule 08.09.2016
comment
Версия спарка 1.6.1.   -  person Leo    schedule 08.09.2016

Ответы (2)


arrow_upward
16
arrow_downward

Попробуйте без Seq:

case class TestPerson(name: String, age: Long, salary: Double)
val tom = TestPerson("Tom Hanks",37,35.5)
val sam = TestPerson("Sam Smith",40,40.5)
val PersonList = mutable.MutableList[TestPerson]()
PersonList += tom
PersonList += sam

val personDS = PersonList.toDS()
println(personDS.getClass)
personDS.show()

val personDF = PersonList.toDF()
println(personDF.getClass)
personDF.show()
personDF.select("name", "age").show()

Вывод:

class org.apache.spark.sql.Dataset

+---------+---+------+
|     name|age|salary|
+---------+---+------+
|Tom Hanks| 37|  35.5|
|Sam Smith| 40|  40.5|
+---------+---+------+

class org.apache.spark.sql.DataFrame

+---------+---+------+
|     name|age|salary|
+---------+---+------+
|Tom Hanks| 37|  35.5|
|Sam Smith| 40|  40.5|
+---------+---+------+

+---------+---+
|     name|age|
+---------+---+
|Tom Hanks| 37|
|Sam Smith| 40|
+---------+---+

Кроме того, не забудьте переместить объявление класса case TestPerson за пределы вашего объекта.

person Ajeet Shah    schedule 08.09.2016
comment
Спасибо за приведенное выше решение, оно сработало в случае набора данных. Моя конечная цель - получить данные в DataFrame. Я использовал эту команду scala› val RowsDF = sc.parallelize(personDS).toDF(), но получил ошибку ‹console›:51: ошибка: несоответствие типов; найдено: org.apache.spark.sql.Dataset[TestPerson] требуется: Seq[?] val RowsDF = sc.parallelize(personDS).toDF() - person Leo; 09.09.2016
comment
Я получил это: scala> val RowsDF = personDS.toDF() RowsDF: org.apache.spark.sql.DataFrame = [имя: строка, возраст: bigint, зарплата: двойная] - person Leo; 09.09.2016

arrow_upward
1
arrow_downward

Используйте последовательность:

val spark = SparkSession.builder().appName("Spark-SQL").master("local[2]").getOrCreate()

import spark.implicits._

var tom = new TestPerson("Tom Hanks",37,35.5)
var sam = new TestPerson("Sam Smith",40,40.5)

val PersonList = mutable.MutableList[TestPerson]()

//Adding data in list
PersonList += tom
PersonList += sam

//It will be work.
var personDS = Seq(PersonList).toDS()

SQLContext.implicits

person zhming    schedule 05.07.2020