Wie subtrahiert man eine Spalte von Tagen von einer Datumsspalte in Pyspark?

Question

die folgende PySpark Datenrahmen

df = sqlContext.createDataFrame([('2015-01-15', 10), 
           ('2015-02-15', 5)], 
           ('date_col', 'days_col')) 

Bedenkt man, wie die Tage Spalte von der Datumsspalte abgezogen werden kann? In diesem Beispiel sollte die resultierende Spalte ['2015-01-05', '2015-02-10'] lauten.

Ich schaute in pyspark.sql.functions.date_sub(), aber es erfordert eine Datumsspalte und einen einzigen Tag, d.h. date_sub(df['date_col'], 10). Im Idealfall würde ich lieber date_sub(df['date_col'], df['days_col']) tun.

Ich habe auch versucht eine UDF zu erstellen:

from datetime import timedelta 
def subtract_date(start_date, days_to_subtract): 
    return start_date - timedelta(days_to_subtract) 

subtract_date_udf = udf(subtract_date, DateType()) 
df.withColumn('subtracted_dates', subtract_date_udf(df['date_col'], df['days_col']) 

Diese technisch funktioniert, aber ich habe gelesen, dass zwischen Funken und Python Schritt Performance-Probleme für große Datensätze verursachen können. Ich kann jetzt bei dieser Lösung bleiben (keine Notwendigkeit, vorzeitig zu optimieren), aber mein Bauchgefühl sagt, dass es einfach eine Möglichkeit geben muss, diese einfache Sache zu tun, ohne eine Python-UDF zu verwenden.

Answer 1

Ich konnte dies mit lösen.

df.selectExpr('date_sub(date_col, day_col) as subtracted_dates') 

Wenn Sie die Spalte auf den ursprünglichen DF anhängen, fügen Sie einfach * zum Ausdruck

df.selectExpr('*', 'date_sub(date_col, day_col) as subtracted_dates') 

Answer 2

nicht die eleganteste Lösung immer, aber wenn Sie nicht wollen, SQL-Ausdrücke in Scala hacken (nicht, dass es schwer sein sollte, aber diese sind privat sql) so etwas wie dies sollte es tun:

from pyspark.sql import Column 

def date_sub_(c1: Column, c2: Column) -> Column: 
    return ((c1.cast("timestamp").cast("long") - 60 * 60 * 24 * c2) 
     .cast("timestamp").cast("date")) 

Für Python 2.x löschen Sie einfach Typ Anmerkungen.

Answer 3

etwas anderes Format, sondern auch funktioniert:

df.registerTempTable("dfTbl") 

newdf = spark.sql(""" 
        SELECT *, date_sub(d.date_col, d.day_col) AS DateSub 
        FROM dfTbl d 
        """)