Spark, Scala - type de colonne de déterminer

Je peux charger des données à partir de la base de données, et je n'certains processus avec ces données.
Le problème est un tableau a, colonne de date en tant que "Chaîne", mais certains autres traits comme 'timestamp'.

Je ne peut pas savoir quel est le type de la colonne de la date jusqu'à ce que le chargement des données.

> x.getAs[String]("date") //could be error when date column is timestamp type
> x.getAs[Timestamp]("date") //could be error when date column is string type

C'est la façon de charger des données à partir de l'étincelle.

spark.read
              .format("jdbc")
              .option("url", url)
              .option("dbtable", table)
              .option("user", user)
              .option("password", password)
              .load()

Est-il possible de trait en même temps? ou de le convertir en chaîne de toujours?

InformationsquelleAutor J.Done | 2016-12-27
  1. 11

    Vous pouvez faire correspondre à un modèle sur le type de la colonne (à l'aide du DataFrame de schéma) pour décider d'analyser la Chaîne dans un Timestamp, ou simplement utiliser le Timestamp qu'est - et de l'utilisation de la unix_timestamp fonction de faire la conversion réelle:

    import org.apache.spark.sql.functions._
import org.apache.spark.sql.types.StringType

//preparing some example data - df1 with String type and df2 with Timestamp type
val df1 = Seq(("a", "2016-02-01"), ("b", "2016-02-02")).toDF("key", "date")
val df2 = Seq(
  ("a", new Timestamp(new SimpleDateFormat("yyyy-MM-dd").parse("2016-02-01").getTime)),
  ("b", new Timestamp(new SimpleDateFormat("yyyy-MM-dd").parse("2016-02-02").getTime))
).toDF("key", "date")

//If column is String, converts it to Timestamp
def normalizeDate(df: DataFrame): DataFrame = {
  df.schema("date").dataType match {
    case StringType => df.withColumn("date", unix_timestamp($"date", "yyyy-MM-dd").cast("timestamp"))
    case _ => df
  }
}

//after "normalizing", you can assume date has Timestamp type - 
//both would print the same thing:
normalizeDate(df1).rdd.map(r => r.getAs[Timestamp]("date")).foreach(println)
normalizeDate(df2).rdd.map(r => r.getAs[Timestamp]("date")).foreach(println)
    InformationsquelleAutor Tzach Zohar
  2. 2

    Voici quelques choses que vous pouvez essayer:

    (1) Commencer d'utiliser le inferSchema fonction au cours de la charge si vous avez une version qui prend en charge. Ce sera l'étincelle figure le type de données des colonnes, ce qui ne fonctionne pas dans tous les scénarios. Regardez également les données d'entrée, si vous avez des citations, je conseille d'ajouter un argument supplémentaire à prendre en compte lors de la charge. 

    val inputDF = spark.read.format("csv").option("header","true").option("inferSchema","true").load(fileLocation)

    (2) Pour identifier le type de données d'une colonne, vous pouvez utiliser le code ci-dessous, il placera le nom de la colonne et des types de données dans leurs propres Tableaux de Chaînes.

    val columnNames : Array[String] = inputDF.columns
val columnDataTypes : Array[String] = inputDF.schema.fields.map(x=>x.dataType).map(x=>x.toString)
    InformationsquelleAutor afeldman
  3. 1

    Il a un moyen facile de remédier à ce qui est get(i: Int): Any. Et ça sera la carte entre Étincelle types SQL et types de retour automatiquement. par exemple,

    val fieldIndex = row.fieldIndex("date")
val date = row.get(fieldIndex)
    InformationsquelleAutor hy-2013
  4. 0
    def parseLocationColumn(df: DataFrame): DataFrame = {
  df.schema("location").dataType match {
    case StringType => df.withColumn("locationTemp", $"location")
      .withColumn("countryTemp", lit("Unknown"))
      .withColumn("regionTemp", lit("Unknown"))
      .withColumn("zoneTemp", lit("Unknown"))
    case _ => df.withColumn("locationTemp", $"location.location")
      .withColumn("countryTemp", $"location.country")
      .withColumn("regionTemp", $"location.region")
      .withColumn("zoneTemp", $"location.zone")
  }
}
    InformationsquelleAutor Rajiv Singh