Comment calculer la date de différence dans pyspark?

J'ai des données comme ceci:

df = sqlContext.createDataFrame([
    ('1986/10/15', 'z', 'null'), 
    ('1986/10/15', 'z', 'null'),
    ('1986/10/15', 'c', 'null'),
    ('1986/10/15', 'null', 'null'),
    ('1986/10/16', 'null', '4.0')],
    ('low', 'high', 'normal'))

Je veux calculer la date de différence entre low colonne et 2017-05-02 et remplacer low colonne avec la différence. J'ai essayé les solutions liées à la sur stackoverflow mais aucun des deux ne fonctionne.

OriginalL'auteur Mark Krystal | 2017-05-17

  1. 19

    Vous avez besoin de jeter la colonne low à la classe date et puis vous pouvez utiliser datediff() en combinaison avec lit(). À L'Aide De Étincelle 2.2:

    from pyspark.sql.functions import datediff, to_date, lit

df.withColumn("test", 
              datediff(to_date(lit("2017-05-02")),
                       to_date("low","yyyy/MM/dd"))).show()
+----------+----+------+-----+
|       low|high|normal| test|
+----------+----+------+-----+
|1986/10/15|   z|  null|11157|
|1986/10/15|   z|  null|11157|
|1986/10/15|   c|  null|11157|
|1986/10/15|null|  null|11157|
|1986/10/16|null|   4.0|11156|
+----------+----+------+-----+

    À l'aide de < Étincelle 2.2, nous avons besoin de convertir le low colonne à la classe timestamp première:

    from pyspark.sql.functions import datediff, to_date, lit, unix_timestamp

df.withColumn("test", 
              datediff(to_date(lit("2017-05-02")),
                       to_date(unix_timestamp('low', "yyyy/MM/dd").cast("timestamp")))).show()
    J'ai une erreur "TypeError: to_date() prend exactement 1 argument (2)"
    Et si il y a Nan valeur dans low de la colonne?
    c'est parce que vous êtes à l'aide de Spark < 2.2
    voir mise à jour avec exemple
    Merci. Le ci-dessus est juste un test de données. Mes données réelles a de nombreuses valeurs low colonne qui ne peuvent pas être jeté dans la timestamp. Comment puis-je mettre ces valeurs à NaN lors de " cast("timestamp") `? homoserine llike pandas: pd.to_datetime(erreurs = 'forcer')

    OriginalL'auteur mtoto

  2. 3

    Sinon, comment faire pour trouver le nombre de jours écoulés entre deux actions de l'utilisateur à l'aide de pySpark:

    import pyspark.sql.functions as funcs
from pyspark.sql.window import Window

window = Window.partitionBy('user_id').orderBy('action_date')

df = df.withColumn("days_passed", funcs.datediff(df.action_date, 
                                  lag(df.action_date, 1).over(window)))



+----------+-----------+-----------+
|   user_id|action_date|days_passed| 
+----------+-----------+-----------+
|623       |2015-10-21|        null|
|623       |2015-11-19|          29|
|623       |2016-01-13|          59|
|623       |2016-01-21|           8|
|623       |2016-03-24|          63|
+----------+----------+------------+

    OriginalL'auteur Artem Zaika