Comment ajouter des colonnes dans org.apache.spark.sql.Ligne à l'intérieur de mapPartitions

Habituellement, il ne devrait pas être nécessaire pour que et il vaut mieux utiliser les fonctions définies par l'utilisateur, mais vous êtes ici:

Comment dois-je définir la fonction de la passer dans df.rdd.mapPartitions, si je veux créer une nouvelle Ligne, avec quelques colonnes supplémentaires

Il devrait prendre Iterator[Row] et retour Iterator[T] donc, dans votre cas, vous devriez utiliser quelque chose comme ceci

import org.apache.spark.sql.Row

def transformRows(iter: Iterator[Row]): Iterator[Row] = ???

Comment puis-je ajouter quelques colonnes dans la Ligne d'objet(ou en créer un nouveau)

Il y a plusieurs manières d'accéder à Row valeurs, y compris Row.get* méthodes, Row.toSeq etc. Nouveau Row peuvent être créés à l'aide de Row.apply, Row.fromSeq, Row.fromTuple ou RowFactory. Par exemple:

def transformRow(row: Row): Row =  Row.fromSeq(row.toSeq ++ Array[Any](-1, 1))

Comment créer DataFrame de créé RDD

Si vous avez RDD[Row] vous pouvez utiliser SQLContext.createDataFrame et de fournir de schéma.

Mettre tous ensemble:

import org.apache.spark.sql.types.{IntegerType, StructField, StructType}

val  df = sc.parallelize(Seq(
    (1.0, 2.0), (0.0, -1.0),
    (3.0, 4.0), (6.0, -2.3))).toDF("x", "y")

def transformRows(iter: Iterator[Row]): Iterator[Row] = iter.map(transformRow)

val newSchema = StructType(df.schema.fields ++ Array(
  StructField("z", IntegerType, false), StructField("v", IntegerType, false)))

sqlContext.createDataFrame(df.rdd.mapPartitions(transformRows), newSchema).show

//+---+----+---+---+
//|  x|   y|  z|  v|
//+---+----+---+---+
//|1.0| 2.0| -1|  1|
//|0.0|-1.0| -1|  1|
//|3.0| 4.0| -1|  1|
//|6.0|-2.3| -1|  1|
//+---+----+---+---+

OriginalL'auteur zero323