Ajouter une colonne à la Trame de Données dans Apache Spark 1.3

Est-il possible et ce qui serait le plus efficace soigné méthode pour ajouter une colonne à la Trame de Données?

Plus précisément, la colonne peut servir comme Id de Ligne pour que la Trame de Données.

Dans un cas simplifié, de la lecture du fichier et non pas à la segmentation, je peux penser à quelque chose comme ci-dessous (en Scala), mais elle les complète avec des erreurs (à la ligne 3), et de toute façon n'a pas l'air comme le meilleur itinéraire possible: 

var dataDF = sc.textFile("path/file").toDF() 
val rowDF = sc.parallelize(1 to DataDF.count().toInt).toDF("ID") 
dataDF = dataDF.withColumn("ID", rowDF("ID"))
  • Avez-vous trouvé une solution?
  • Quelles sont les erreurs? Cela semble être la bonne méthode dans l'API.
  • withColumn est censé fonctionner avec la même dataframe - dire que vous pouvez faire quelque chose comme dataDF = dataDF.withColumn("ID", dataDF("ID").map(...)) avec la colonne de this dataframe seulement, pas avec d'autres
  • ah. hein. ce serait bien si cette restriction ont été documentés. Problème intéressant ensuite. le seul mécanisme que je peux voir à partir de l'API, autrement, serait d'utiliser un join. Pas un mauvais candidat à suggérer pour une amélioration de l'API.
  • à droite, c'est l'un des principaux funcitonalities pour n'importe quel bloc de données. Par exemple, ils ont des fonctionnalités intégrées pour mettre à jour le schéma de Parquet fichiers. Évidemment, c'est très cher fonctionnement dans un environnement distribué, mais de toute façon de l'OMI, il devrait être là
  • Avez-vous essayé d'utiliser un UDF? (quelque chose comme sqlContext.udf().register("...

InformationsquelleAutor Oleg Shirokikh | 2015-04-07
  1. 50

    Il a été un moment depuis que j'ai posté la question et il semble que d'autres personnes aimeraient avoir une réponse ainsi. Ci-dessous est ce que j'ai trouvé.

    De sorte que la tâche d'origine a été d'ajouter une colonne avec la ligne identificators (en gros, une séquence 1 to numRows) à un bloc de données, de sorte que les lignes de commande/de présence peuvent être suivis (par exemple, lorsque vous de l'échantillon). Ceci peut être atteint par quelque chose le long de ces lignes:

    sqlContext.textFile(file).
zipWithIndex().
map(case(d, i)=>i.toString + delimiter + d).
map(_.split(delimiter)).
map(s=>Row.fromSeq(s.toSeq))

    Qui concerne le cas général, l'ajout d'une colonne à tout bloc de données:

    Le "plus proche" de cette fonctionnalité dans l'Étincelle de l'API sont withColumn et withColumnRenamed. Selon Scala docs, l'ancien Retourne une nouvelle DataFrame par l'ajout d'une colonne. À mon avis, c'est un peu confus et incomplet définition. Ces deux fonctions peuvent être utilisées sur this bloc de données seulement, soit, compte tenu de deux trames de données df1 et df2 avec la colonne col:

    val df = df1.withColumn("newCol", df1("col") + 1) //-- OK
val df = df1.withColumn("newCol", df2("col") + 1) //-- FAIL

    Donc, sauf si vous pouvez gérer pour transformer une colonne dans un dataframe de la forme dont vous avez besoin, vous ne pouvez pas utiliser withColumn ou withColumnRenamed pour l'ajout arbitraire colonnes (autonome ou d'autres blocs de données).

    Comme il a été dit ci-dessus, la solution de contournement solution peut être d'utiliser un join - ce serait assez compliqué, bien que possible, - fixer les clés uniques, comme ci-dessus avec zipWithIndex à la fois des trames de données ou des colonnes pourrait fonctionner. Bien que l'efficacité est ...

    Il est clair que l'ajout d'une colonne pour le bloc de données n'est pas facile de fonctionnalités pour l'environnement distribué et il peut ne pas être très efficace, propre méthode pour que tout. Mais je pense que c'est toujours très important d'avoir ces fonctionnalités de base, même avec des performances mises en garde.

    • Il y a certaines récente initiative de soutien à la fonction row_number (entre autres) - SPARK-7712 Le ticket JIRA ne le mentionne pas explicitement, de sorte que vous voudrez peut-être regarder dans le liées pull-request
    • Bonne réponse! L'ajout de la colonne fonctionnalité peut être optimisé si l'Étincelle connu que je vais rejoindre sur le triés clé. Que serait un booster de performance.
    • Si vous cherchez à ajouter une colonne id vous devriez regarder la fonction monotonically_increasing_id (), qui peut être utilisé à l'intérieur de withColumn.
    • monotonically_increasing_id() a une profonde fragilité questions et vous devez être très prudent si vous utilisez stackoverflow.com/questions/35705038/...
    InformationsquelleAutor Oleg Shirokikh
  2. 27

    ne sais pas si il fonctionne spark 1.3 mais dans l'étincelle de 1,5-je utiliser withColumn:

    import sqlContext.implicits._
import org.apache.spark.sql.functions._


df.withColumn("newName",lit("newValue"))

    Je l'utiliser quand j'ai besoin d'utiliser une valeur qui n'est pas liée à des colonnes existantes de la dataframe

    Ceci est similaire à @NehaM de la réponse, mais plus simple

    InformationsquelleAutor Tal Joffe
  3. 6

    J'ai pris l'aide de la réponse ci-dessus. Cependant, je le trouve incomplet si nous voulons changer un DataFrame et actuel Api sont peu différentes dans Spark 1.6.
    zipWithIndex() renvoie une Tuple de (Row, Long) qui contient chaque ligne et l'index correspondant. Nous pouvons l'utiliser pour créer de nouveaux Row selon notre besoin. 

    val rdd = df.rdd.zipWithIndex()
             .map(indexedRow => Row.fromSeq(indexedRow._2.toString +: indexedRow._1.toSeq))
val newstructure = StructType(Seq(StructField("Row number", StringType, true)).++(df.schema.fields))
sqlContext.createDataFrame(rdd, newstructure ).show

    J'espère que ce sera utile.

    InformationsquelleAutor NehaM
  4. 3

    Vous pouvez utiliser row_number avec Fonction fenêtre comme ci-dessous pour obtenir les différents id de chaque côté des lignes dans un dataframe.

    df.withColumn("ID", row_number() over Window.orderBy("any column name in the dataframe"))

    Vous pouvez également utiliser monotonically_increasing_id pour la même chose que 

    df.withColumn("ID", monotonically_increasing_id())

    Et il y a quelques d'autres moyens trop.

    InformationsquelleAutor Ramesh Maharjan