flinksql+维表

Flink SQL中的维表JOIN是指将流式数据与外部数据源中的维表进行关联查询，为实时计算提供数据关联。在维表JOIN时，需要指明这条记录关联维表快照的时刻。需要注意的是，目前Flink SQL的维表JOIN仅支持对当前时刻维表快照的关联(处理时间语义)，而不支持事实表rowtime所对应的维表快照(事件时间语义)。

下面是一个使用Flink SQL进行维表JOIN的例子：

假设我们有一个订单表order_table和一个商品表product_table，其中订单表中包含商品ID，我们需要将订单表中的商品ID关联到商品表中获取商品名称和价格等信息。这时我们可以使用维表JOIN来实现。

首先，我们需要在Flink SQL中创建订单表和商品表：

CREATE TABLE order_table (
  order_id BIGINT,
  product_id BIGINT,
  order_time TIMESTAMP(3),
  WATERMARK FOR order_time AS order_time - INTERVAL '5' SECOND
) WITH (
  'connector.type' = 'kafka',
  'connector.version' = 'universal',
  'connector.topic' = 'order_topic',
  'connector.startup-mode' = 'latest-offset',
  'connector.properties.zookeeper.connect' = 'localhost:2181',
  'connector.properties.bootstrap.servers' = 'localhost:9092',
  'format.type' = 'json'
);

CREATE TABLE product_table (
  product_id BIGINT,
  product_name STRING,
  price DECIMAL(10, 2),
  update_time TIMESTAMP(3),
  WATERMARK FOR update_time AS update_time - INTERVAL '5' SECOND
) WITH (
  'connector.type' = 'jdbc',
  'connector.url' = 'jdbc:mysql://localhost:3306/test',
  'connector.table' = 'product',
  'connector.driver' = 'com.mysql.jdbc.Driver',
  'connector.username' = 'root',
  'connector.password' = '123456',
  'connector.lookup.cache.max-rows' = '5000',
  'connector.lookup.cache.ttl' = '10s'
);

其中，order_table是从Kafka中读取的订单数据，product_table是从MySQL中读取的商品数据。

接下来，我们可以使用维表JOIN将订单表和商品表进行关联查询：

SELECT o.order_id, o.product_id, p.product_name, p.price
FROM order_table AS o
JOIN product_table FOR SYSTEM_TIME AS OF o.order_time AS p
ON o.product_id = p.product_id;

在这个例子中，我们使用了FOR SYSTEM_TIME AS OF子句来指定关联维表的时刻为订单表中的订单时间。这样，我们就可以在Flink SQL中实现维表JOIN了。