Flink Lookup Join（维表 Join）

🕗 发布于 2024-11-20 15:49 flink linq 大数据

Lookup Join 定义（支持 Batch\Streaming）

Lookup Join 其实就是维表 Join，比如拿离线数仓来说，常常会有用户画像，设备画像等数据，而对应到实时数仓场景中，这种实时获取外部缓存的 Join 就叫做维表 Join。

应用场景：

Lookup Join 是流与 Redis，Mysql，HBase 这种存储介质的 Join。Lookup 的意思就是实时查找，而实时的画像数据一般都是存储在 Redis，Mysql，HBase 中，这就是 Lookup Join 的由来；

实际案例

kafka流表和mysql维表的关联：
使用曝光用户日志流（show_log）关联用户画像维表（user_profile）关联到用户的维度之后，提供给下游计算分性别，年龄段的曝光用户数使用。

mysql端处理：

[root@spop007~]# mysql -uroot -p123456


mysql> create database test;
mysql> CREATE TABLE `user_profile` (
  `user_id` varchar(100) NOT NULL,
  `age` varchar(100) DEFAULT NULL,
  `sex` varchar(100) DEFAULT NULL,
  PRIMARY KEY (`user_id`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8;
INSERT INTO test.user_profile (user_id,age,sex) VALUES
 ('a','12-18','男'),
 ('b','18-24','女'),
 ('c','18-24','男');

mysql>select * from test.user_profile;

kafka端处理：

# 1.创建Kafka主题 test_k，指定分区数量为1，副本数量为1
kafka-topics.sh \
--create \
--topic test_k \
--bootstrap-server localhost:9092 \
--partitions 1 \
--replication-factor 1


# 2.向 test_k 中写入JSON格式的样例数据
./kafka-console-producer.sh \
--topic test_k \
--bootstrap-server localhost:9092

!!!!!这个错误是因为你使用的 Kafka 版本较旧，不支持 --bootstrap-server 参数。旧版本的 Kafka 使用
参数代替 --broker-list 
----------------------------------
./kafka-console-producer.sh \
--topic test_k \
--broker-list localhost:9092
-----------------------------------
#输入完上面脚本，直接粘贴复制json
{"log_id": "1", "timestamp": "1635696063","user_id":"a"}
{"log_id": "2", "timestamp": "1635696180","user_id":"b"}
{"log_id": "3", "timestamp": "1635696300","user_id":"c"}
{"log_id": "4", "timestamp": "1635696360","user_id":"b"}
{"log_id": "5", "timestamp": "1635696420","user_id":"c"}
{"log_id": "6", "timestamp": "1635696420","user_id":"d"}

# 3.创建一个消费者组 group_k1 来消费 test_k 数据
kafka-console-consumer.sh \
--topic test_k \
--bootstrap-server localhost:9092 \
--group group_k1 \
--from-beginning

Flinksql代码：

前提：

jdbc的jar包和mysql的驱动包，都需要事先放入$FLINK_HOME/lib目录下。

flink-connector-jdbc-1.15.2.jar

mysql-connector-java-8.0.29.jar

cd $FLINK_HOME/bin
./sql-client.sh    

CREATE TABLE click_log_table (
  log_id BIGINT, 
  `timestamp` bigint,
  user_id string,
  proctime AS PROCTIME()
) WITH (
    'connector' = 'kafka',
    'topic' = 'test_k',
    'properties.bootstrap.servers' = '192.168.77.88:9092',
    'properties.group.id' = 'group_k1',
    'scan.startup.mode' = 'earliest-offset',
    'format' = 'json'
);

 CREATE TABLE user_profile (
  `user_id` string, 
  `age` string,
  `sex` string
)
WITH (
   'connector' = 'jdbc',
   'url' = 'jdbc:mysql://192.168.77.88:3306/test',
   'table-name' = 'user_profile',
   'username'='root',
   'password'='root'
);

SELECT 
    s.log_id as log_id
    , s.`timestamp` as `timestamp`
    , s.user_id as user_id
    , s.proctime as proctime
    , u.sex as sex
    , u.age as age
FROM click_log_table AS s
LEFT JOIN user_profile FOR SYSTEM_TIME AS OF s.proctime AS u
ON s.user_id = u.user_id;


查看flinksql输出窗口显示：


               log_id            timestamp                        user_id                proctime                            sex                            age
                    1           1635696063                              a 2024-11-19 00:28:14.404                              男                          12-18
                    2           1635696180                              b 2024-11-19 00:28:14.407                              女                          18-24
                    3           1635696300                              c 2024-11-19 00:28:14.409                              男                          18-24
                    4           1635696360                              b 2024-11-19 00:28:14.412                              女                          18-24
                    5           1635696420                              c 2024-11-19 00:28:14.422                              男                          18-24
                    6           1635696420                              d 2024-11-19 00:28:14.424                         (NULL)                         (NULL)

在这里插入图片描述

修改mysql的数据 查看动态表的变化
UPDATE user_profile
SET age = '99-99', sex = 0
WHERE user_id = "a";


kafka端输入：
{"log_id": "11111111111", "timestamp": "1635696063","user_id":"a"}
结果对应下图一

kafka端再输入：
{"log_id": "222222", "timestamp": "1635696063","user_id":"a"}
结果对应下图二