CREATE TABLE kafka_source (
  log_time STRING,
  log_ts AS TO_TIMESTAMP(log_time,'yyyy-MM-dd HH:mm:ss.SSS'),
  user_id INT,
  behavior STRING,
  proctime as PROCTIME(),
  WATERMARK FOR log_ts AS log_ts - INTERVAL '10' SECOND
) WITH (
  'connector' = 'kafka',
  'topic' = 'eventLogJson',
  'properties.bootstrap.servers' = 'kafka_nodeip:9093',
  'properties.group.id' = 'testGroup',
  'scan.startup.mode' = 'latest-offset',
  'format' = 'json',
  'properties.security.protocol' = 'SASL_PLAINTEXT',   
  'properties.sasl.mechanism' = 'SCRAM-SHA-512',   
  'properties.sasl.jaas.config' = 'org.apache.flink.kafka.shaded.org.apache.kafka.common.security.scram.ScramLoginModule required username="xxx" password="***";'
);
CREATE TABLE mysql_user_profile (
  user_id INT,
  city STRING
) WITH (
   'connector' = 'jdbc',
   'url' = 'jdbc:mysql://mysql_ip:3306/biz_online?useSSL=false',
   'table-name' = 'user_profile',
   'username' = 'xxx',
   'password' = '***',
   'lookup.cache' = 'PARTIAL',
   'lookup.cache.max-rows' = '1000',
   'lookup.partial-cache.expire-after-write' = '60 min'
);
CREATE CATALOG edap_catalog WITH ('type' = 'edap');
INSERT INTO edap_catalog.`default`.test_hive_sink_bos_datalake
/*+OPTIONS('partition.time-extractor.timestamp-pattern'='$day $hour:00:00', 
'sink.partition-commit.trigger'='partition-time', 
'sink.partition-commit.delay'='5 min', 
'sink.partition-commit.policy.kind'='metastore,success-file')*/
SELECT ods.user_id, behavior, dim.city, DATE_FORMAT(log_ts, 'yyyy-MM-dd'), DATE_FORMAT(log_ts, 'HH') 
FROM kafka_source as ods
LEFT JOIN mysql_user_profile FOR SYSTEM_TIME AS OF ods.proctime as dim ON ods.user_id = dim.user_id;