session_window
Génère une fenêtre de session étant donné une colonne spécifiant le Timestamp.
La fenêtre de session est l’une des fenêtres dynamiques, ce qui signifie que la longueur de la fenêtre varie en fonction des entrées données. La durée de la fenêtre de session est définie comme « le Timestamp de la dernière entrée de la session + durée de l'intervalle », de sorte que lorsque les nouvelles entrées sont liées à la fenêtre de session actuelle, l'heure de fin de la fenêtre de session peut être étendue en fonction des nouvelles entrées.
Windows peut prendre en charge la précision à la microseconde. Windows dans l'ordre des mois ne sont pas prises en charge.
Pour une query de streaming, vous pouvez utiliser la fonction current_timestamp pour générer des fenêtres sur le temps de traitement.
gapDuration est fourni sous forme de chaînes, par ex. '1 seconde', '1 jour 12 heures', '2 minutes'. Les chaînes d'intervalle valides sont « week », « day », « hour », « minute », « second », « millisecond », « microsecond ».
Il pourrait aussi s'agir d'une colonne pouvant être évaluée dynamiquement en durée de l'écart en fonction de la ligne d'entrée.
La colonne de sortie sera une structure appelée « session_window » par default, avec les colonnes imbriquées « start » et « end », où « start » et « end » seront de pyspark.sql.types.TimestampType.
Pour la fonction Databricks SQL correspondante, consultez l'expression de regroupementsession_window.
Syntaxe
from pyspark.sql import functions as dbf
dbf.session_window(timeColumn=<timeColumn>, gapDuration=<gapDuration>)
parameter
parameter | Type | Description |
|---|---|---|
|
| Le nom de la colonne ou la colonne à utiliser comme Timestamp pour le fenêtrage temporel. La colonne d'horodatage doit être de type TimestampType ou TimestampNTZType. |
|
| Un littéral de chaîne Python ou une colonne spécifiant le délai d'expiration de la session. Il pourrait s'agir d'une valeur statique, par exemple. |
Renvoie
pyspark.sql.Column: la colonne pour les résultats calculés.
Exemples
from pyspark.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()