Kumulative Summe einer Partition in PySpark

Question

Ich muss eine Spalte mit einer Gruppennummer erstellen, die basierend auf den Werten in der Spalte TRUE inkrementiert. Ich kann nach ID partitionieren, also denke ich, dass dies das Inkrement zurücksetzen würde, wenn sich die ID ändert, was ich tun möchte. Innerhalb der ID möchte ich die Gruppennummer immer dann erhöhen, wenn TRUE ungleich 1 ist. Wenn TRUE = 1, möchte ich, dass die Nummer die gleiche wie die letzte bleibt. Dies ist eine Teilmenge meiner aktuellen ID- und TRUE-Spalten, und GROUP wird wie gewünscht angezeigt. Ich habe auch Spalten LATITUDE und LONGITUDE, die ich in meiner Art verwende.

ID TRUE GROUP 
3828 0 1 
3828 0 2 
3828 1 2 
3828 1 2 
3828 1 2 
4529 0 1 
4529 1 1 
4529 0 2 
4529 1 2 
4529 0 3 
4529 0 4 
4529 1 4 
4529 0 5 
4529 1 5 
4529 1 5 

Ich hatte gehofft, so etwas wie unten zu tun, aber das gibt mir alle 0s

trip.registerTempTable("trip_temp") 
trip2 = sqlContext.sql('select *, sum(cast(TRUE = 0 as int)) over(partition by ID order by ID, LATITUDE, LONGITUDE) as GROUP from trip_temp') 

Answer 1

Nie begrenzenden Schlüsselwörter als Spaltennamen verwenden. Auch wenn dies in einigen Systemen funktionieren könnte, ist es fehleranfällig, kann nicht mehr funktionieren, wenn Sie den Parser ändern und im Allgemeinen eine schlechte Übung ist. TRUE ist boolean wörtlichen und wird nie gleich 0 (mit impliziter Besetzung ist es gleichbedeutend mit TRUE IS NOT TRUE)

spark.createDataFrame(
    [(3828, 0, 1), (3828, 1, 2)], ("ID", "TRUE", "GROUP") 
).createOrReplaceTempView("trip_temp") 

spark.sql("SELECT TRUE = 0 AS foo FROM trip_temp LIMIT 2").show() 

// +-----+ 
// | foo| 
// +-----+ 
// |false| 
// |false| 
// +-----+ 

Wenn Sie es wirklich Gebrauch Backticks machen will arbeiten:

spark.sql("SELECT `TRUE` = 0 AS foo FROM trip_temp LIMIT 2").show() 

// +-----+ 
// | foo| 
// +-----+ 
// | true| 
// |false| 
// +-----+ 

aber bitte don‘ t.

Answer 2

Ich weiß, die Frage ist ziemlich alt. Ich wollte nur für diejenigen, die Sie suchen nach einem optimalen Weg zu teilen.

from pyspark.sql.window import * 
import sys 

cumSumPartition = Window.partitionBy(['col1','col2','col3','col4']).orderBy("col3").rowsBetween(-sys.maxsize -1,0) 

temp = temp.withColumn("col5",sum(temp.col5).over(cumSumPartition))