セッションウィンドウ
タイムスタンプ指定列を指定してセッション ウィンドウを生成します。
セッション ウィンドウは動的ウィンドウの 1 つであり、指定された入力に応じてウィンドウの長さが変化することを意味します。セッション ウィンドウの長さは、「セッションの最新の入力のタイムスタンプ + ギャップ期間」として定義されるため、新しい入力が現在のセッション ウィンドウにバインドされると、セッション ウィンドウの終了時刻は新しい入力に応じて拡張できます。
Windows はマイクロ秒の精度をサポートできます。月順のWindowsはサポートされていません。
ストリーミング クエリの場合、関数current_timestampを使用して処理時間のウィンドウを生成できます。gapDurationは文字列として提供されます。例:「1秒」、「1日12時間」、「2分」。有効な間隔文字列は、「week」、「day」、「hour」、「minute」、「second」、「millisecond」、「microsecond」です。
また、入力行に基づいてギャップ期間を動的に評価できる列にすることもできます。
出力列は、デフォルトではネストされた列を持つ「session_window」という構造体になります。
「start」と「end」。「start」と「end」はpyspark.sql.types.TimestampTypeになります。
対応する Databricks SQL 関数については、 session_windowグループ化式を参照してください。
構文
from pyspark.databricks.sql import functions as dbf
dbf.session_window(timeColumn=<timeColumn>, gapDuration=<gapDuration>)
パラメーター
パラメーター | Type | 説明 |
|---|---|---|
|
| 時間によるウィンドウ処理のタイムスタンプとして使用する列名または列。時間列は TimestampType または TimestampNTZType である必要があります。 |
|
| セッションのタイムアウトを指定する Python 文字列リテラルまたは列。静的な値である可能性がある。例： |
戻り値
pyspark.sql.Column: コンピュート結果の列。
例
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()