在flink sql中进行时态表连接时如何设置hbase事件时间答案

【问题标题】：how to set hbase event-time when make a temporal table join in flink sql在flink sql中进行时态表连接时如何设置hbase事件时间
【发布时间】：2021-12-21 23:23:50
【问题描述】：

left 是来自 kafka 的事实表

CREATE TABLE  dig_user_join_kafka (
id string,
username string,
city_id string,
create_time TIMESTAMP(3),
WATERMARK FOR create_time AS create_time - INTERVAL '5' SECOND

)

right 是来自 hbase 的维度表

CREATE TABLE dim_city_hbase (
 id string,
 info ROW＜
name string
＞,
 PRIMARY KEY (id) NOT ENFORCED
)

我想与事件时间进行临时表连接

insert into dim_city_join_hbase 
select id as id, 
ROW(username, city, create_time) as info 
from ( 
select kj.id as id, 
kj.username as username, 
hj.info.name as city, 
kj.create_time as create_time 
from dig_user_join2_kafka kj
left join dim_city_hbase FOR SYSTEM_TIME AS OF kj.create_time  hj
 on kj.city_id = hj.id
)

现在，错误是

The main method caused an error: Event-Time Temporal Table Join requires both primary
key and row time attribute in versioned table, but no row time attribute can be found

表示hbase表没有行时间，如何设置hbase event-time？

许多示例显示 hbase temporal table join with proctime，但没有人使用 event-time，

【问题讨论】：

标签： join apache-flink temporal

【解决方案1】：

使用系统函数“PROCTIME()”或“CURRENT_ROW_TIMESTAMP()”添加一个字段作为行时间然后设置水印：

" update_time AS CURRENT_ROW_TIMESTAMP(), " +
" WATERMARK FOR update_time AS update_time, " +

【讨论】：