Meilleur moyen de convertir un champ de chaîne en horodatage dans Spark

J'ai un fichier CSV dans lequel un champ datetime dans un format spécifique. Je ne peux pas l'importer directement dans mon Dataframe, parce qu'il faut un horodatage. J'ai donc l'importer en tant que chaîne et de le convertir en un Timestamp comme ce

import java.sql.Timestamp
import java.text.SimpleDateFormat
import java.util.Date
import org.apache.spark.sql.Row

def getTimestamp(x:Any) : Timestamp = {
    val format = new SimpleDateFormat("MM/dd/yyyy' 'HH:mm:ss")
    if (x.toString() == "") 
    return null
    else {
        val d = format.parse(x.toString());
        val t = new Timestamp(d.getTime());
        return t
    }
}

def convert(row : Row) : Row = {
    val d1 = getTimestamp(row(3))
    return Row(row(0),row(1),row(2),d1)
}

Est-il mieux, de manière plus concise à cela, le Dataframe de l'API ou de l'étincelle-sql? La méthode ci-dessus nécessite la création d'un EDR et à donner le schéma du Dataframe de nouveau.

source d'informationauteur user568109

Spark >= 2.2

import org.apache.spark.sql.functions.to_timestamp

val ts = to_timestamp($"dts", "MM/dd/yyyy HH:mm:ss")

df.withColumn("ts", ts).show(2, false)

//+---+-------------------+-------------------+
//|id |dts                |ts                 |
//+---+-------------------+-------------------+
//|1  |05/26/2016 01:01:01|2016-05-26 01:01:01|
//|2  |#$@#@#             |null               |
//+---+-------------------+-------------------+

Spark >= 1.6, < 2.2

Vous pouvez utiliser la date de la transformation des fonctions qui ont été introduites dans Spark 1.5. En supposant que vous avez les données suivantes:

val df = Seq((1L, "05/26/2016 01:01:01"), (2L, "#$@#@#")).toDF("id", "dts")

Vous pouvez utiliser unix_timestamp analyser les chaînes, et de le jeter à l'horodatage

import org.apache.spark.sql.functions.unix_timestamp

val ts = unix_timestamp($"dts", "MM/dd/yyyy HH:mm:ss").cast("timestamp")

df.withColumn("ts", ts).show(2, false)

//+---+-------------------+---------------------+
//|id |dts                |ts                   |
//+---+-------------------+---------------------+
//|1  |05/26/2016 01:01:01|2016-05-26 01:01:01.0|
//|2  |#$@#@#             |null                 |
//+---+-------------------+---------------------+

Comme vous pouvez le voir, il couvre à la fois l'analyse et le traitement des erreurs.

Spark >= 1.5, < 1.6

Vous aurez à utiliser quelque chose comme ceci:

unix_timestamp($"dts", "MM/dd/yyyy HH:mm:ss").cast("double").cast("timestamp")

(unix_timestamp($"dts", "MM/dd/yyyy HH:mm:ss") * 1000).cast("timestamp")

en raison de SPARK-11724.

Spark < 1.5

vous devriez être en mesure de les utiliser avec expr et HiveContext.

6

Je n'ai pas joué avec Spark SQL encore, mais je pense que ce serait plus idiomatiques scala (null utilisation n'est pas considérée comme une bonne pratique):
```
def getTimestamp(s: String) : Option[Timestamp] = s match {
  case "" => None
  case _ => {
    val format = new SimpleDateFormat("MM/dd/yyyy' 'HH:mm:ss")
    Try(new Timestamp(format.parse(s).getTime)) match {
      case Success(t) => Some(t)
      case Failure(_) => None
    }    
  }
}
```
Remarquez s'il vous plaît je suppose que vous savez Row de types d'éléments à l'avance (si vous l'avez lu à partir d'un fichier csv, elles sont String), c'est pourquoi j'ai utiliser un type comme String et pas Any (tout est sous-type de Any).

Elle dépend aussi de la façon dont vous souhaitez gérer les exceptions de parsage. Dans ce cas, si une analyse exception se produit, un None est simplement retourné.

Vous pouvez l'utiliser plus loin avec:
```
rows.map(row => Row(row(0),row(1),row(2), getTimestamp(row(3))
```

J'ai ISO8601 timestamp dans mon dataset et j'avais besoin de le convertir en "aaaa-MM-jj". C'est ce que j'ai fait:

import org.joda.time.{DateTime, DateTimeZone}
object DateUtils extends Serializable {
  def dtFromUtcSeconds(seconds: Int): DateTime = new DateTime(seconds * 1000L, DateTimeZone.UTC)
  def dtFromIso8601(isoString: String): DateTime = new DateTime(isoString, DateTimeZone.UTC)
}

sqlContext.udf.register("formatTimeStamp", (isoTimestamp : String) => DateUtils.dtFromIso8601(isoTimestamp).toString("yyyy-MM-dd"))

Et vous pouvez simplement utiliser l'UDF dans votre étincelle de requête SQL.

Je voudrais déplacer le getTimeStamp méthode écrit par vous en rdd est mapPartitions et la réutilisation GenericMutableRow entre les lignes dans un itérateur:

val strRdd = sc.textFile("hdfs://path/to/cvs-file")
val rowRdd: RDD[Row] = strRdd.map(_.split('\t')).mapPartitions { iter =>
  new Iterator[Row] {
    val row = new GenericMutableRow(4)
    var current: Array[String] = _

    def hasNext = iter.hasNext
    def next() = {
      current = iter.next()
      row(0) = current(0)
      row(1) = current(1)
      row(2) = current(2)

      val ts = getTimestamp(current(3))
      if(ts != null) {
        row.update(3, ts)
      } else {
        row.setNullAt(3)
      }
      row
    }
  }
}

Et vous devriez toujours utiliser le schéma pour générer un DataFrame

val df = sqlContext.createDataFrame(rowRdd, tableSchema)

L'utilisation de GenericMutableRow à l'intérieur d'un itérateur de la mise en œuvre pourrait se trouver dans Agrégation De L'Opérateur InMemoryColumnarTableScan ParquetTableOperations etc.

Je voudrais utiliser https://github.com/databricks/spark-csv

Cela permettra de déduire des horodateurs pour vous.

import com.databricks.spark.csv._
val rdd: RDD[String] = sc.textFile("csvfile.csv")

val df : DataFrame = new CsvParser().withDelimiter('|')
      .withInferSchema(true)
      .withParseMode("DROPMALFORMED")
      .csvRdd(sqlContext, rdd)

Vous devez vous connecter pour publier un commentaire.