Bdtjtk

Scala noob here. Still trying to learn the syntax.

I am trying to reduce the code I have to write to convert my test data into DataFrames. Here is what I have right now:

  def makeDf[T](seq: Seq[(Int, Int)], colNames: String*): Dataset[Row] = {

    val context = session.sqlContext

    import context.implicits._

    seq.toDF(colNames: _*)

  }

The problem is that the above method only takes a sequence of the shape Seq[(Int, Int)] as input. How do I make it take any sequence as input? I can change the inputs shape to Seq[AnyRef], but then the code fails to recognize the toDF call as valid symbol.

I am not able to figure out how to make this work. Any ideas? Thanks!

Short answer:

import scala.reflect.runtime.universe.TypeTag



def makeDf[T <: Product: TypeTag](seq: Seq[T], colNames: String*): DataFrame = ...

Explanation:

When you are calling seq.toDF you are actually using an implicit defined in SQLImplicits:

implicit def localSeqToDatasetHolder[T : Encoder](s: Seq[T]): DatasetHolder[T] = {

  DatasetHolder(_sqlContext.createDataset(s))

}

which in turn requires the generation of an encoder. The problem is that encoders are defined only on certain types. Specifically Product (i.e. tuple, case class etc.) You also need to add the TypeTag implicit so that Scala can get over the type erasure (in the runtime all Sequences have the type sequence regardless of the generics type. TypeTag provides information on this).

As a side node, you do not need to extract sqlcontext from the session, you can simply use:

import sparkSession.implicits._

As @AssafMendelson already explained the real reason of why you cannot create a Dataset of Any is because Spark needs an Encoder to transform objects from they JVM representation to its internal representation - and Spark cannot guarantee the generation of such Encoder for Any type.

Assaf answers is correct, and will work.

However, IMHO, it is too much restrictive as it will only work for Products (tuples, and case classes) - and even if that includes most use cases, there still a few ones excluded.

Since, what you really need is an Encoder, you may leave that responsibility to the client. Which in most situation will only need to call import spark.implicits._ to get them in scope.

Thus, this is what I believe will be the most general solution.

import org.apache.spark.sql.{DataFrame, Dataset, Encoder, SparkSession}



// Implicit SparkSession to make the call to further methods more transparent.

implicit val spark = SparkSession.builder.master("local[*]").getOrCreate()

import spark.implicits._



def makeDf[T: Encoder](seq: Seq[T], colNames: String*)

                      (implicit spark: SparkSession): DataFrame =

  spark.createDataset(seq).toDF(colNames: _*)



def makeDS[T: Encoder](seq: Seq[T])

                      (implicit spark: SparkSession): Dataset[T] =

  spark.createDataset(seq)

Note: This is basically re-inventing the already defined functions from Spark.