apache-spark
Spark DataFrame

Introduction

Création de DataFrames dans Scala

Il existe plusieurs façons de créer des DataFrames. Ils peuvent être créés à partir de listes locales, de RDD distribués ou de lectures à partir de sources de données.

Utiliser toDF

En important les implicits spark sql, on peut créer un DataFrame à partir d'un Seq, Array ou RDD local, tant que le contenu est d'un sous-type Product (tuples et classes de cas sont des exemples bien connus de sous-types Product). Par exemple:

import sqlContext.implicits._
val df = Seq(
  (1, "First Value", java.sql.Date.valueOf("2010-01-01")),
  (2, "Second Value", java.sql.Date.valueOf("2010-02-01"))
).toDF("int_column", "string_column", "date_column")

Utiliser createDataFrame

Une autre option consiste à utiliser la méthode createDataFrame présente dans SQLcontext. Cette option permet également la création à partir de listes locales ou de RDD de sous-types de produits comme avec toDF , mais les noms des colonnes ne sont pas définis dans la même étape. Par exemple:

val df1 = sqlContext.createDataFrame(Seq(
  (1, "First Value", java.sql.Date.valueOf("2010-01-01")),
  (2, "Second Value", java.sql.Date.valueOf("2010-02-01"))
))

En outre, cette approche permet la création de fichiers RDD d'instances de Row , à condition qu'un paramètre de schema soit transmis pour la définition du schéma de DataFrame résultant. Exemple:

import org.apache.spark.sql.types._
val schema = StructType(List(
    StructField("integer_column", IntegerType, nullable = false),
    StructField("string_column", StringType, nullable = true),
    StructField("date_column", DateType, nullable = true)
))

val rdd = sc.parallelize(Seq(
  Row(1, "First Value", java.sql.Date.valueOf("2010-01-01")),
  Row(2, "Second Value", java.sql.Date.valueOf("2010-02-01"))
))

val df = sqlContext.createDataFrame(rdd, schema)

Lecture de sources

Le moyen le plus courant de créer DataFrame provient peut-être des sources de données. On peut le créer à partir d'un fichier parquet en hdfs, par exemple:

val df = sqlContext.read.parquet("hdfs:/path/to/file")