Temporal Table Function
Temporal Table Function
Temporal table function은 특정 시점에서 temporal table(시간에 따라 변하는 테이블)의 버전에 접근할 수 있게 해줍니다. temporal table에서 데이터에 접근하려면 반환될 테이블의 버전을 결정하는 time attribute를 전달해야 합니다. Flink는 이를 표현하는 방법으로 table functions의 SQL 구문을 사용합니다.
versioned table과 달리 temporal table functions은 append-only 스트림 위에서만 정의할 수 있습니다. changelog 입력을 지원하지 않습니다. 또한 temporal table function은 순수 SQL DDL로는 정의할 수 없습니다.
출처: 문서
본문
Temporal Table Function 정의하기
Temporal table functions은 Table API를 사용해 append-only 스트림 위에 정의할 수 있습니다. 테이블은 하나 이상의 키 열과 버저닝에 사용되는 time attribute로 등록됩니다.
temporal table function으로 등록하려는 통화 환율의 append-only 테이블이 있다고 가정해 봅시다.
SELECT * FROM currency_rates;
update_time currency rate
============= ========= ====
09:00:00 Yen 102
09:00:00 Euro 114
09:00:00 USD 1
11:15:00 Euro 119
11:49:00 Pounds 108
Table API를 사용해 currency를 키로, update_time을 버저닝 time attribute로 이 스트림을 등록할 수 있습니다.
Java
TemporalTableFunction rates = tEnv
.from("currency_rates")
.createTemporalTableFunction("update_time", "currency");
tEnv.createTemporarySystemFunction("rates", rates);
Scala
rates = tEnv
.from("currency_rates")
.createTemporalTableFunction("update_time", "currency")
tEnv.createTemporarySystemFunction("rates", rates)
Python
Still not supported in Python API.
Temporal Table Function Join
정의되면 temporal table function은 표준 table function으로 사용됩니다. append-only 테이블(왼쪽 입력/probe 측)은 시간에 따라 변하며 그 변경을 추적하는 temporal table(오른쪽 입력/build 측)과 조인하여 특정 시점의 키 값을 검색할 수 있습니다.
고객의 다양한 통화 주문을 추적하는 append-only 테이블 orders를 생각해 봅시다.
SELECT * FROM orders;
order_time amount currency
========== ====== =========
10:15 2 Euro
10:30 1 USD
10:32 50 Yen
10:52 3 Euro
11:04 5 USD
이 테이블들이 주어졌을 때 주문을 공통 통화인 USD로 변환하려고 합니다.
SQL
SELECT
SUM(amount * rate) AS amount
FROM
orders,
LATERAL TABLE (rates(order_time))
WHERE
rates.currency = orders.currency
Java
Table result = orders
.joinLateral(call("rates", $("o_proctime")), $("o_currency").isEqual($("r_currency")))
.select($("(o_amount").times($("r_rate")).sum().as("amount"));
Scala
val result = orders
.joinLateral($"rates(order_time)", $"orders.currency = rates.currency")
.select($" (o_amount * r_rate).sum as amount"))
Python
Still not supported in Python API.