session_window

Generuje okno sesji przy użyciu znacznika czasu określającego kolumnę.

Okno sesji jest jednym z okien dynamicznych, co oznacza, że długość okna różni się w zależności od danych wejściowych. Długość okna sesji jest definiowana jako "sygnatura czasowa najnowszych danych wejściowych sesji i czasu trwania przerwy", więc gdy nowe dane wejściowe są powiązane z bieżącym oknem sesji, czas zakończenia okna sesji można rozszerzyć zgodnie z nowymi danymi wejściowymi.

System Windows może obsługiwać precyzję mikrosekund. System Windows w kolejności miesięcy nie jest obsługiwany.

W przypadku zapytania przesyłania strumieniowego można użyć funkcji current_timestamp do generowania okien w czasie przetwarzania. funkcja gapDuration jest dostarczana jako ciągi, np. "1 sekunda", "1 dzień 12 godzin", "2 minuty". Prawidłowe ciągi interwału to "tydzień", "dzień", "godzina", "minuta", "sekunda", "milisekunda", "mikrosekunda".

Może to być również kolumna, która może być obliczana w celu dynamicznego trwania przerwy na podstawie wiersza wejściowego.

Kolumna wyjściowa będzie domyślnie strukturą o nazwie "session_window" z zagnieżdżonych kolumn "start" i "end", gdzie wartości "start" i "end" będą mieć wartość pyspark.sql.types.TimestampType.

Aby uzyskać odpowiednią funkcję SQL usługi Databricks, zobacz session_window wyrażenie grupowania.

Składnia

from pyspark.databricks.sql import functions as dbf

dbf.session_window(timeColumn=<timeColumn>, gapDuration=<gapDuration>)

Parametry

Parameter	Typ	Description
timeColumn	pyspark.sql.Column lub str	Nazwa kolumny lub kolumna do użycia jako sygnatura czasowa dla okien według czasu. Kolumna czasowa musi mieć wartość TimestampType lub TimestampNTZType.
gapDuration	pyspark.sql.Column lub literal string	Literał lub kolumna ciągu języka Python określająca limit czasu sesji. Może to być wartość statyczna, np. 10 minutes, 1 secondlub wyrażenie/funkcja UDF, która określa czas trwania luki dynamicznie na podstawie wiersza wejściowego.

Zwraca

pyspark.sql.Column: kolumna dla obliczonych wyników.

Przykłady

from pyspark.databricks.sql import functions as dbf
df = spark.createDataFrame([('2016-03-11 09:00:07', 1)], ['dt', 'v'])
df2 = df.groupBy(dbf.session_window('dt', '5 seconds')).agg(dbf.sum('v'))
df2.show(truncate=False)
df2.printSchema()