FlinkSQL流表与维表join和双流join
维表是数仓中的⼀个概念，维表中的维度属性是观察数据的⾓度，补充事实表的信息。在实时数仓中，同样也有维表与事实表的概念，其中事实表通常为kafka的实时流数据，维表通常存储在外部设备中(⽐如MySQL，HBase)。对于每条流式数据，可以关联⼀个外部维表数据源，为实时计算提供数据关联查询。维表可能是会不断变化的，在维表JOIN时，需指明这条记录关联维表快照的时刻。
本⽂主要介绍：
1、流表和维表的区别
2、流表和维表join的数据流转解析
3、双流join的数据流转解析
4、代码⽰例和场景
1、流表和维表的区别：
流表：实时流数据映射成的表，在join查询中，每来⼀条数据都会主动去维表中查询是否有匹配的数据
维表：维度信息表，⼀般在外部存储（Redis，Mysql中）维表是被动查询的，⽬前Flink SQL的维表JOIN仅⽀持对当前时刻维表快照的关联
2、流表和维表join：
3、流表和流表关联（双流jion）：
4、代码⽰例（FlinkSQL）
流表和维表join
场景：
流表是⽤户的⾏为数据，记录⽤户学习内容，学习时长等信息
维表是学⽣信息数据（姓名，⼿机号）
如果cpt表有⼀条实时流数据A同学的学习记录过来，此时student表还没有这个学⽣的信息，获取到的姓名和⼿机号就是null。当student 表更新有A的信息，后续的实时流数据能关联获取到A的姓名和⼿机号，但是student表更新之前的实时流数据依旧是获取不到的（这点需要注意）
创建kafka数据源，流表
ute_sql(
"""CREATE TABLE cpt(
content_id string,
student_uid string,
course_id string,
content_type int,
study_time bigint,
chapter_id string,
section_id string,
progress int,
updated_at string,
created_at string,
playback_time int,
ptime AS PROCTIME()
) WITH (
'connector' = 'kafka',
'topic' = 'stream_content_study_progress',
'properties.bootstrap.servers' = 'localhost:9092',
'up.id' = 't1',
'de'='earliest-offset',
'format' = 'json'
)
""")
创建Redis维表
ute_sql(
"""create table student(
uid string,
mobile string,
name string,
PRIMARY KEY (uid) NOT ENFORCED
PRIMARY KEY (uid) NOT ENFORCED
)with(
'connector'='redis',
'ip'='localhost',
'password'='',
'port'='6379',
'table'='user',
'database'='1'
)
""")
创建sink表
ute_sql("""create table res(
uid string,
mobile string,
courseId string,
viewingTime bigint,
updateTime bigint
)with(
'connector'='kafka',
'topic'='course_viewing_v1',
'properties.bootstrap.servers' = 'localhost:9092',
'format' = 'changelog-json'
)""")
流表和维表join
ute_sql(
"""
insert into res
select
t.student_uid as uid,
t.study_time as viewingTime,
unix_timestamp(t.updated_at)*1000 as updateTime,
from cpt as t
left join student  for system_time as of t.ptime on t.student_uid=student.uid
"""
)
流表和流表join
场景：上⾯描述的场景，通过双流join能很好的解决。两个都是流表，两个表当中的任何⼀个表来数据时，都会去另⼀个表中查询是否有匹配的数据（包含历史）
创建流表
ute_sql(
"""CREATE TABLE cpt(
content_id string,
student_uid string,
course_id string,
content_type int,
study_time bigint,
chapter_id string,
section_id string,
progress int,
updated_at string,
created_at string,
playback_time int,
ptime AS PROCTIME()
) WITH (
'connector' = 'kafka',
'topic' = 'stream_content_study_progress',
'properties.bootstrap.servers' = 'localhost:9092',          'up.id' = 't1',
'de'='earliest-offset',
'format' = 'json'
)
""")connect和join的区别
创建流表，其实是维度表
ute_sql(
"""create table student(
uid string,
mobile string,
name string,
PRIMARY KEY (uid) NOT ENFORCED
)with(
'connector' = 'kafka',
'topic' = 'stream_student',
'properties.bootstrap.servers' = 'localhost:9092',          'up.id' = 't1',
'de'='earliest-offset',
'format' = 'json'
)
双流join
ute_sql(
"""
insert into res
select
t.student_uid as uid,
t.study_time as viewingTime,
unix_timestamp(t.updated_at)*1000 as updateTime,    from cpt as t
left join student  t.student_uid=student.uid
"""
)

版权声明：本站内容均来自互联网，仅供演示用，请勿用于商业和其他非法用途。如果侵犯了您的权益请与我们联系QQ:729038198，我们将在24小时内删除。