janela_de_sessão
Gera uma janela de sessão a partir de um carimbo de data/hora que especifica a coluna.
A janela de sessão é uma janela dinâmica, o que significa que seu tamanho varia de acordo com as entradas fornecidas. A duração da janela de sessão é definida como "o registro de data e hora da última entrada da sessão + a duração do intervalo", portanto, quando novas entradas são vinculadas à janela de sessão atual, o horário de término da janela de sessão pode ser expandido de acordo com as novas entradas.
O Windows suporta precisão em microssegundos. Não há suporte Windows na ordem dos meses.
Para uma consulta de transmissão, você pode usar a função current_timestamp para gerar janelas de tempo de processamento.
A duração do intervalo é fornecida como strings, por exemplo. '1 segundo', '1 dia e 12 horas', '2 minutos'. strings de intervalo válidas são 'semana', 'dia', 'hora', 'minuto', 'segundo', 'milissegundo' e 'microssegundo'.
Também pode ser uma coluna que pode ser avaliada dinamicamente para determinar a duração do intervalo com base na linha de entrada.
A coluna de saída será uma estrutura chamada 'session_window' por default , com as colunas aninhadas.
'começar' e 'fim', onde 'começar' e 'fim' serão de pyspark.sql.types.TimestampType.
Para a função Databricks SQL correspondente, consulte a expressão de agrupamentosession_window.
Sintaxe
from pyspark.databricks.sql import functions as dbf
dbf.session_window(timeColumn=<timeColumn>, gapDuration=<gapDuration>)
Parâmetros
Parâmetro | Tipo | Descrição |
|---|---|---|
|
| O nome da coluna ou a coluna a ser usada como carimbo de data/hora para o recorte por tempo. A coluna de tempo deve ser do tipo TimestampType ou TimestampNTZType. |
|
| Uma string literal Python ou uma coluna que especifica o tempo limite da sessão. Pode ser um valor estático, por exemplo. |
Devoluções
pyspark.sql.Column: a coluna para resultados de cálculo.
Exemplos
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()