SESI - Realtime Compute for Apache Flink

Topik ini menjelaskan penggunaan fungsi SESI di Realtime Compute for Apache Flink.

Definisi

Fungsi SESI mengelompokkan elemen berdasarkan aktivitas sesi. Tidak seperti jendela tumbling dan sliding, jendela sesi tidak tumpang tindih dan ukurannya tidak tetap. Jika tidak ada elemen baru yang diterima dalam periode waktu tertentu, sesi akan terputus dan jendela ditutup.

Sintaksis

Anda dapat menggunakan fungsi SESI dalam klausa GROUP BY untuk mendefinisikan jendela sesi.

SESI(<time-attr>, <gap-interval>)

Parameter input

Parameter	Deskripsi	Contoh
time-attr	Parameter ini harus berupa bidang atribut waktu yang valid dalam sebuah aliran. Parameter ini menentukan apakah waktu tersebut adalah waktu pemrosesan atau waktu kejadian. Untuk informasi lebih lanjut, lihat Atribut waktu.	-
gap-interval	Periode timeout atau interval inaktivitas untuk sesi. Jika tidak ada elemen baru yang tiba dalam periode waktu yang ditentukan oleh `<gap-interval>` setelah elemen terakhir dari sesi tiba, sesi ditutup. Elemen yang tiba kemudian dialokasikan ke sesi baru. Parameter gap-interval dalam format `INTERVAL 'num' timeUnit`.	`INTERVAL '10' SECOND` menunjukkan bahwa periode timeout sesi adalah 10 detik.

Fungsi pengidentifikasi jendela

Fungsi pengidentifikasi jendela menentukan waktu mulai, waktu akhir, atau atribut waktu dari sebuah jendela. Atribut waktu digunakan untuk menggabungkan jendela tingkat lebih rendah.

Fungsi	Tipe nilai kembali	Deskripsi
`SESSION_START (<time-attr>, <gap-interval>)`	Timestamp	Mengembalikan waktu mulai, termasuk nilai batas, dari sebuah jendela. Sebagai contoh, jika rentang waktu jendela adalah `[00:10, 00:15]`, `00:10` dikembalikan. Nilai kembali adalah waktu dari catatan pertama dalam jendela sesi.
`SESSION_END (<time-attr>, <gap-interval>)`	Timestamp	Mengembalikan waktu akhir, termasuk nilai batas, dari sebuah jendela. Sebagai contoh, jika rentang waktu jendela adalah `[00:00, 00:15]`, `00:15` dikembalikan. Nilai kembali adalah waktu dari catatan terakhir dalam jendela sesi ditambah `<gap-interval>`.
`SESSION_ROWTIME (<time-attr>, <gap-interval>)`	TIMESTAMP (rowtime-attr)	Mengembalikan waktu akhir, tidak termasuk nilai batas, dari sebuah jendela. Sebagai contoh, jika rentang waktu jendela adalah `(00:00, 00:15)`, `00:14:59.999` dikembalikan. Nilai kembali adalah atribut rowtime berdasarkan mana operasi waktu dapat dilakukan. Fungsi ini hanya dapat digunakan dalam jendela yang didefinisikan berdasarkan waktu kejadian, seperti jendela bertingkat. Untuk informasi lebih lanjut, lihat Jendela bertingkat. Fungsi ini hanya berlaku untuk jendela berbasis waktu kejadian.
`SESSION_PROCTIME(<time-attr>, <gap-interval>)`	TIMESTAMP (rowtime-attr)	Mengembalikan waktu akhir, tidak termasuk nilai batas, dari sebuah jendela. Sebagai contoh, jika rentang waktu jendela adalah `(00:00, 00:15)`, `00:14:59.999` dikembalikan. Nilai kembali adalah atribut waktu pemrosesan berdasarkan mana operasi waktu dapat dilakukan. Fungsi ini hanya dapat digunakan dalam jendela yang didefinisikan berdasarkan waktu pemrosesan, seperti jendela bertingkat. Untuk informasi lebih lanjut, lihat Jendela bertingkat. Fungsi ini hanya berlaku untuk jendela berbasis waktu pemrosesan.

Contoh

Contoh berikut menunjukkan cara menghitung jumlah klik per pengguna selama setiap sesi aktif. Interval timeout sesi adalah 30 detik.

Data uji pada tabel user_clicks

username (VARCHAR)	click_url (VARCHAR)	eventtime (VARCHAR)
Jark	`http://taobao.com/xxx`	`2024-10-10 10:00:00.0`
Jark	`http://taobao.com/xxx`	`2024-10-10 10:00:10.0`
Jark	`http://taobao.com/xxx`	`2024-10-10 10:00:49.0`
Jark	`http://taobao.com/xxx`	`2024-10-10 10:01:05.0`
Jark	`http://taobao.com/xxx`	`2024-10-10 10:01:58.0`
Timo	`http://taobao.com/xxx`	`2024-10-10 10:02:10.0`

Pernyataan uji

CREATE TEMPORARY TABLE user_clicks(
  username varchar,
  click_url varchar,
  eventtime varchar,                            
  ts AS TO_TIMESTAMP(eventtime),
  WATERMARK FOR ts AS ts - INTERVAL '2' SECOND  -- Tentukan watermark untuk rowtime.
) WITH (
  'connector' = 'kafka',
  'topic' = '<yourTopic>',
  'properties.bootstrap.servers' = '<brokers>',
  'scan.startup.mode' = 'earliest-offset',
  'format' = 'csv'
);

CREATE TEMPORARY TABLE session_output(
  window_start TIMESTAMP,
  window_end TIMESTAMP,
  username VARCHAR,
  clicks BIGINT
) WITH (
  'connector'='print',
  'logger'='true'
);

INSERT INTO session_output
SELECT
SESSION_START(ts, INTERVAL '30' SECOND),
SESSION_END(ts, INTERVAL '30' SECOND),
username,
COUNT(click_url)
FROM user_clicks
GROUP BY SESSION(ts, INTERVAL '30' SECOND), username;

Hasil uji

window_start (TIMESTAMP)	window_end (TIMESTAMP)	username (VARCHAR)	clicks (BIGINT)
`2024-10-10 10:00:00.0`	`2024-10-10 10:00:40.0`	Jark	2
`2024-10-10 10:00:49.0`	`2024-10-10 10:01:35.0`	Jark	2
`2024-10-10 10:01:58.0`	`2024-10-10 10:02:28.0`	Jark	1
`2024-10-10 10:02:10.0`	`2024-10-10 10:02:40.0`	Timo	1