Différence entre l'API DataSet et l'API DataFrame

Question

Est-ce que quelqu'un peut m'aider à comprendre la différence entre l'API DataSet et l'API DataFrame avec un exemple ? Pourquoi était-il nécessaire d'introduire l'API DataSet dans Spark ?

Answer 1

Apache Spark fournit trois types d'APIs

1. RDD
2. DataFrame
3. Ensemble de données

Voici la comparaison des API entre RDD, Dataframe et Dataset.

RDD

La principale abstraction fournie par Spark est un ensemble de données distribuées résilientes (RDD), c'est-à-dire une collection d'éléments répartis entre les nœuds du cluster et pouvant être exploités en parallèle.

Caractéristiques du RDD : -

• Collection distribuée :
  RDD utilise les opérations MapReduce qui sont largement adoptées pour traiter et générer de grands ensembles de données avec un algorithme parallèle et distribué sur un cluster. Il permet aux utilisateurs d'écrire des calculs parallèles, en utilisant un ensemble d'opérateurs de haut niveau, sans avoir à se soucier de la distribution du travail et de la tolérance aux pannes.

• Immuable : Les RDDs sont composés d'une collection d'enregistrements qui sont partitionnés. Une partition est une unité de base du parallélisme dans un RDD, et chaque partition est une division logique des données qui est immuable et créée par des transformations sur des partitions existantes.

• Tolérance aux pannes : Dans le cas où nous perdons une partition du RDD, nous pouvons rejouer la transformation sur cette partition dans la lignée pour réaliser le même calcul, plutôt que de répliquer les données sur plusieurs nœuds. Cette caractéristique est le plus grand avantage du RDD car elle permet d'économiser beaucoup d'efforts dans la gestion et la réplication des données et donc de réaliser des calculs plus rapides.

• Des évaluations paresseuses : Toutes les transformations dans Spark sont paresseuses, en ce sens qu'elles ne calculent pas leurs résultats tout de suite. Au lieu de cela, elles se souviennent simplement des transformations appliquées à un ensemble de données de base. Les transformations ne sont calculées que lorsqu'une action nécessite un résultat à renvoyer au programme pilote.

• Transformations fonctionnelles : Les RDD prennent en charge deux types d'opérations : les transformations, qui créent un nouvel ensemble de données à partir d'un ensemble existant, et les actions, qui renvoient une valeur au programme pilote après avoir effectué un calcul sur l'ensemble de données.

• Formats de traitement des données :
  Il peut traiter facilement et efficacement les données, qu'elles soient structurées ou non.

• Langages de programmation supportés :
  L'API RDD est disponible en Java, Scala, Python et R.

Limites de RDD:-

• Pas de moteur d'optimisation intégré : Lorsqu'on travaille avec des données structurées, les RDD ne peuvent pas tirer parti des optimiseurs avancés de Spark, notamment l'optimiseur de catalyseur et le moteur d'exécution Tungsten. Les développeurs doivent optimiser chaque RDD en fonction de ses attributs.

• Manipulation de données structurées : Contrairement aux Dataframe et aux ensembles de données, les RDD ne déduisent pas le schéma des données ingérées et nécessitent que l'utilisateur le spécifie.

Cadres de données

Spark a introduit les Dataframes dans la version 1.3 de Spark. Les Dataframes permettent de surmonter les principaux problèmes rencontrés par les RDD.

Un DataFrame est une collection distribuée de données organisées en colonnes nommées. Il est conceptuellement équivalent à une table dans une base de données relationnelle ou à un Dataframe R/Python. Avec le Dataframe, Spark a également introduit l'optimiseur de catalyseur, qui tire parti de fonctionnalités de programmation avancées pour construire un optimiseur de requête extensible.

Caractéristiques de Dataframe:-

• Collection distribuée de Row Objects : Un DataFrame est une collection distribuée de données organisées en colonnes nommées. Il est conceptuellement équivalent à une table dans une base de données relationnelle, mais avec des optimisations plus riches sous le capot.

• Traitement des données : Traitement des formats de données structurées et non structurées (Avro, CSV, elastic search et Cassandra) et des systèmes de stockage (HDFS, tables HIVE, MySQL, etc.). Il peut lire et écrire à partir de toutes ces diverses sources de données.

• Optimisation à l'aide de l'optimiseur de catalyseur : Il alimente à la fois les requêtes SQL et l'API DataFrame. Dataframe utilise un cadre de transformation d'arbre catalyseur en quatre phases,

  1.Analyzing a logical plan to resolve references
  2.Logical plan optimization
  3.Physical planning
  4.Code generation to compile parts of the query to Java bytecode.

• Compatibilité avec les ruches : En utilisant Spark SQL, vous pouvez exécuter des requêtes Hive non modifiées sur vos entrepôts Hive existants. Il réutilise le front-end et le MetaStore de Hive et vous offre une compatibilité totale avec les données, les requêtes et les UDF Hive existants.

• Tungstène : Tungsten fournit un backend d'exécution physique qui gère explicitement la mémoire et génère dynamiquement le bytecode pour l'évaluation des expressions.

• Langages de programmation supportés :
  L'API Dataframe est disponible en Java, Scala, Python et R.

Limites du cadre de données : -

• Sécurité des types au moment de la compilation : Comme nous l'avons vu, l'API Dataframe ne prend pas en charge la sécurité au moment de la compilation, ce qui vous limite dans la manipulation des données lorsque la structure n'est pas connue. L'exemple suivant fonctionne au moment de la compilation. Cependant, vous obtiendrez une exception Runtime lors de l'exécution de ce code.

Exemple :

case class Person(name : String , age : Int) 
val dataframe = sqlContext.read.json("people.json") 
dataframe.filter("salary > 10000").show 
=> throws Exception : cannot resolve 'salary' given input age , name

C'est un défi, surtout lorsque vous travaillez avec plusieurs étapes de transformation et d'agrégation.

• Impossible d'opérer sur l'objet de domaine (objet de domaine perdu) : Une fois que vous avez transformé un objet de domaine en dataframe, vous ne pouvez pas le régénérer à partir de celui-ci. Dans l'exemple suivant, une fois que nous avons créé personDF à partir de personRDD, nous ne pourrons plus récupérer le RDD original de la classe Person (RDD[Person]).

Exemple :

case class Person(name : String , age : Int)
val personRDD = sc.makeRDD(Seq(Person("A",10),Person("B",20)))
val personDF = sqlContext.createDataframe(personRDD)
personDF.rdd // returns RDD[Row] , does not returns RDD[Person]

API de jeux de données

L'API Dataset est une extension des DataFrames qui fournit une interface de programmation orientée objet et à sécurité intrinsèque. Il s'agit d'une collection d'objets immuables, fortement typés, qui sont mis en correspondance avec un schéma relationnel.

Au cœur de l'API Dataset, on trouve un nouveau concept appelé encodeur, qui est responsable de la conversion entre les objets JVM et la représentation tabulaire. La représentation tabulaire est stockée en utilisant le format binaire Tungsten interne à Spark, ce qui permet d'effectuer des opérations sur des données sérialisées et d'améliorer l'utilisation de la mémoire. Spark 1.6 est livré avec un support pour la génération automatique d'encodeurs pour une grande variété de types, y compris les types primitifs (par exemple String, Integer, Long), les classes de cas Scala, et Java Beans.

Caractéristiques de l'ensemble de données:-

• Fournit le meilleur des deux RDD et Dataframe : RDD (programmation fonctionnelle, sécurité des types), DataFrame (modèle relationnel, optimisation des requêtes, exécution Tungsten, tri et mélange).

• Encodeurs : Grâce aux encodeurs, il est facile de convertir n'importe quel objet de la JVM en un ensemble de données, ce qui permet aux utilisateurs de travailler avec des données structurées et non structurées, contrairement aux cadres de données.

• Langages de programmation supportés : L'API Datasets est actuellement disponible uniquement en Scala et Java. Python et R ne sont actuellement pas pris en charge dans la version 1.6. La prise en charge de Python est prévue pour la version 2.0.

• Sécurité de type : L'API Datasets offre une sécurité au moment de la compilation qui n'était pas disponible dans les Dataframes. Dans l'exemple ci-dessous, nous pouvons voir comment Dataset peut opérer sur des objets de domaine avec des fonctions lambda compilées.

Exemple :

case class Person(name : String , age : Int)
val personRDD = sc.makeRDD(Seq(Person("A",10),Person("B",20)))
val personDF = sqlContext.createDataframe(personRDD)
val ds:Dataset[Person] = personDF.as[Person]
ds.filter(p => p.age > 25)
ds.filter(p => p.salary > 25)
 // error : value salary is not a member of person
ds.rdd // returns RDD[Person]

• Interopérable : Datasets vous permet de convertir facilement vos RDDs et Dataframes existants en datasets sans code passe-partout.

Limitation de l'API Datasets:-

• Nécessite un casting de type vers String : L'interrogation des données à partir d'ensembles de données nécessite actuellement de spécifier les champs de la classe sous forme de chaîne. Une fois que nous avons interrogé les données, nous sommes obligés de convertir la colonne dans le type de données requis. D'autre part, si nous utilisons l'opération map sur les datasets, elle n'utilisera pas l'optimiseur Catalyst.

Exemple :

ds.select(col("name").as[String], $"age".as[Int]).collect()

Pas de prise en charge de Python et de R : à partir de la version 1.6, Datasets ne prend en charge que Scala et Java. La prise en charge de Python sera introduite dans Spark 2.0.

L'API Datasets présente plusieurs avantages par rapport à l'API RDD et Dataframe existante, notamment une meilleure sécurité de type et une programmation fonctionnelle. Avec le défi que représente le moulage de type dans l'API, vous n'obtiendrez toujours pas la sécurité de type requise, ce qui rendra votre code fragile.

Answer 2

Parce que DataFrame est faiblement typé et les développeurs ne bénéficient pas des avantages du système de types. Par exemple, disons que vous voulez lire quelque chose depuis SQL et exécuter une agrégation dessus :

val people = sqlContext.read.parquet("...")
val department = sqlContext.read.parquet("...")

people.filter("age > 30")
  .join(department, people("deptId") === department("id"))
  .groupBy(department("name"), "gender")
  .agg(avg(people("salary")), max(people("age")))

Quand vous dites people("deptId") vous ne récupérez pas une Int ou un Long vous récupérez un Column objet sur lequel vous devez opérer. Dans les langages dotés d'un riche système de types, comme Scala, on finit par perdre toute la sécurité des types, ce qui augmente le nombre d'erreurs d'exécution pour des choses qui pourraient être découvertes au moment de la compilation.

Au contraire, DataSet[T] est tapé. Quand vous le faites :

val people: People = val people = sqlContext.read.parquet("...").as[People]

Vous récupérez en fait un People où deptId est un type intégral réel et non un type de colonne, tirant ainsi parti du système de types.

À partir de la version 2.0 de Spark, les API DataFrame et DataSet seront unifiées. DataFrame sera un alias de type pour DataSet[Row] .