Spark: Ajouter une colonne à l'image de données conditionnellement

J'essaie de prendre mes données d'entrée:

A    B       C
--------------
4    blah    2
2            3
56   foo     3

Et ajouter une colonne à la fin, selon si B est vide ou pas:

A    B       C     D
--------------------
4    blah    2     1
2            3     0
56   foo     3     1

Je peux le faire facilement par l'enregistrement de l'entrée dataframe comme une table temporaire, puis en tapant une requête SQL.

Mais j'aimerais vraiment savoir comment le faire avec Scala de méthodes et de ne pas avoir à taper une requête SQL au sein de la Scala.

J'ai essayé .withColumnmais je n'arrive pas à faire ce que je veux.

source d'informationauteur mcmcmc

  1. 52

    Essayer withColumn avec la fonction when comme suit:

    val sqlContext = new SQLContext(sc)
import sqlContext.implicits._ //for `toDF` and $""
import org.apache.spark.sql.functions._ //for `when`

val df = sc.parallelize(Seq((4, "blah", 2), (2, "", 3), (56, "foo", 3), (100, null, 5)))
    .toDF("A", "B", "C")

val newDf = df.withColumn("D", when($"B".isNull or $"B" === "", 0).otherwise(1))

    newDf.show() montre

    +---+----+---+---+
|  A|   B|  C|  D|
+---+----+---+---+
|  4|blah|  2|  1|
|  2|    |  3|  0|
| 56| foo|  3|  1|
|100|null|  5|  0|
+---+----+---+---+

    J'ai ajouté le (100, null, 5) ligne pour l'examen de la isNull cas.

    J'ai essayé ce code avec Spark 1.6.0 mais comme commenté dans le code de whenil fonctionne sur les versions après 1.4.0.

  2. 2

    My bad, j'avais raté une partie de la question.

    Meilleur, le plus propre est d'utiliser un UDF.
    Explication dans le code.

    //create some example data...BY DataFrame
//note, third record has an empty string
case class Stuff(a:String,b:Int)
val d= sc.parallelize(Seq( ("a",1),("b",2),
     ("",3) ,("d",4)).map { x => Stuff(x._1,x._2)  }).toDF

//now the good stuff.
import org.apache.spark.sql.functions.udf
//function that returns 0 is string empty 
val func = udf( (s:String) => if(s.isEmpty) 0 else 1 )
//create new dataframe with added column named "notempty"
val r = d.select( $"a", $"b", func($"a").as("notempty") )

    scala> r.show
+---+---+--------+
|  a|  b|notempty|
+---+---+--------+
|  a|  1|    1111|
|  b|  2|    1111|
|   |  3|       0|
|  d|  4|    1111|
+---+---+--------+
  3. 0

    Comment quelque chose comme cela? 

    val newDF = df.filter($"B" === "").take(1) match {
  case Array() => df
  case _ => df.withColumn("D", $"B" === "")
}

    À l'aide de take(1) doit avoir un minimum de frapper