Modifier la propriété nullable de la colonne dans la structure de données spark

Je suis création manuelle d'un dataframe pour certains tests. Le code pour créer c'est:

case class input(id:Long, var1:Int, var2:Int, var3:Double)
val inputDF = sqlCtx
  .createDataFrame(List(input(1110,0,1001,-10.00),
    input(1111,1,1001,10.00),
    input(1111,0,1002,10.00)))

De sorte que le schéma ressemble à ceci:

root
 |-- id: long (nullable = false)
 |-- var1: integer (nullable = false)
 |-- var2: integer (nullable = false)
 |-- var3: double (nullable = false)

Je veux faire "nullable = true' pour chacune de ces variables. Comment dois-je déclarer que, depuis le début ou le basculer dans une nouvelle dataframe après, il a été créé?

source d'informationauteur J Calbreath

  1. 19

    Réponse

    Avec les importations

    import org.apache.spark.sql.types.{StructField, StructType}
import org.apache.spark.sql.{DataFrame, SQLContext}
import org.apache.spark.{SparkConf, SparkContext}

    vous pouvez utiliser

    /**
 * Set nullable property of column.
 * @param df source DataFrame
 * @param cn is the column name to change
 * @param nullable is the flag to set, such that the column is  either nullable or not
 */
def setNullableStateOfColumn( df: DataFrame, cn: String, nullable: Boolean) : DataFrame = {

  //get schema
  val schema = df.schema
  //modify [[StructField] with name `cn`
  val newSchema = StructType(schema.map {
    case StructField( c, t, _, m) if c.equals(cn) => StructField( c, t, nullable = nullable, m)
    case y: StructField => y
  })
  //apply new schema
  df.sqlContext.createDataFrame( df.rdd, newSchema )
}

    directement.

    Vous pouvez également faire la méthode disponible via le "pimp my library" (bibliothèque de modèle ( voir mon post Quelle est la meilleure façon de définir des méthodes personnalisées sur un DataFrame? ), de sorte que vous pouvez appeler 

    val df = ....
val df2 = df.setNullableStateOfColumn( "id", true )

    Modifier

    Solution Alternative 1

    Utiliser une légère version modifiée de setNullableStateOfColumn

    def setNullableStateForAllColumns( df: DataFrame, nullable: Boolean) : DataFrame = {
  //get schema
  val schema = df.schema
  //modify [[StructField] with name `cn`
  val newSchema = StructType(schema.map {
    case StructField( c, t, _, m) ⇒ StructField( c, t, nullable = nullable, m)
  })
  //apply new schema
  df.sqlContext.createDataFrame( df.rdd, newSchema )
}

    Solution Alternative 2

    Explicitement définir le schéma. (Utiliser la réflexion pour créer une solution qui est plus général) 

    configuredUnitTest("Stackoverflow.") { sparkContext =>

  case class Input(id:Long, var1:Int, var2:Int, var3:Double)

  val sqlContext = new SQLContext(sparkContext)
  import sqlContext.implicits._


  //use this to set the schema explicitly or
  //use refelection on the case class member to construct the schema
  val schema = StructType( Seq (
    StructField( "id", LongType, true),
    StructField( "var1", IntegerType, true),
    StructField( "var2", IntegerType, true),
    StructField( "var3", DoubleType, true)
  ))

  val is: List[Input] = List(
    Input(1110, 0, 1001,-10.00),
    Input(1111, 1, 1001, 10.00),
    Input(1111, 0, 1002, 10.00)
  )

  val rdd: RDD[Input] =  sparkContext.parallelize( is )
  val rowRDD: RDD[Row] = rdd.map( (i: Input) ⇒ Row(i.id, i.var1, i.var2, i.var3))
  val inputDF = sqlContext.createDataFrame( rowRDD, schema ) 

  inputDF.printSchema
  inputDF.show()
}
  2. 10

    C'est une réponse tardive, mais je voulais donner une solution alternative pour les gens qui viennent ici. Vous pouvez automatiquement faire une DataFrame Column nullable dès le départ par la suite à la modification de votre code:

    case class input(id:Option[Long], var1:Option[Int], var2:Int, var3:Double)
val inputDF = sqlContext
  .createDataFrame(List(input(Some(1110),Some(0),1001,-10.00),
    input(Some(1111),Some(1),1001,10.00),
    input(Some(1111),Some(0),1002,10.00)))
inputDF.printSchema

    Cela donnera:

    root
 |-- id: long (nullable = true)
 |-- var1: integer (nullable = true)
 |-- var2: integer (nullable = false)
 |-- var3: double (nullable = false)

defined class input
inputDF: org.apache.spark.sql.DataFrame = [id: bigint, var1: int, var2: int, var3: double]

    Essentiellement, si vous déclarez un champ comme un Option en utilisant Some([element]) ou None comme les entrées, alors que le champ d'accepter les valeurs null. Sinon, le champ ne sera pas les valeurs null. J'espère que cela aide!

  3. 3

    Plus compact version de réglage de toutes les colonnes nullable paramètre

    Au lieu de case StructField( c, t, _, m) ⇒ StructField( c, t, nullable = nullable, m) on peut utiliser _.copy(nullable = nullable). Puis l'ensemble de la fonction peut être écrite comme suit:

    def setNullableStateForAllColumns( df: DataFrame, nullable: Boolean) : DataFrame = {
  df.sqlContext.createDataFrame(df.rdd, StructType(df.schema.map(_.copy(nullable = nullable))))
}
  4. 1

    Suffit d'utiliser java.lang.Entiers à la place de la scala.Int dans votre classe de cas.

    case class input(id:Long, var1:java.lang.Integer , var2:java.lang.Integer , var3:java.lang.Double)
  5. 1

    Une autre option, si vous avez besoin de changer dataframe en place, et la recréation est impossible, vous pouvez faire quelque chose comme ceci:

    .withColumn("col_name", when(col("col_name").isNotNull, col("col_name")).otherwise(lit(null)))

    Étincelle va alors penser que cette colonne peut contenir des nullet la possibilité de valeur null sera mis à true.
    Aussi, vous pouvez utiliser udf, pour envelopper vos valeurs dans Option.
    Fonctionne très bien, même pour le streaming des cas.