维表Join尝鲜
什么是维表
维表，维度表的简称，来源于数据仓库，一般用来给事实数据补充信息。假设现在有一张销售记录表。销售记录表里面的一条销售记录就是一条事实数据，而这条销售记录中的地区字段就是一个维度。通常销售记录表里面的地区字段是地区表的主键，地区表就是一张维表。更多的细节可以面向百度/谷歌编程。
为什么Flink中需要维表
以流计算为例，一般情况下，消费的消息中间件中的消息，是事实表中的数据，我们需要把数据补全，不然谁也不知道字段地区对应的值01、02是个什么东西。所以，我们通常会在计算过程中，通过Join维表来补全数据。
Flink如何Join维表
1.9之前版本
- 如果我们使用的是DataStream，那直接通过Alibaba贡献给社区的异步I/O，完成维表的连接
- 如果是Flink Sql，那么需要先通过
calcite将sql翻译成DataStream，再通过异步I/O，完成维表关联 - 如果直接通过
FlatMap之类的算子，在open()中连接第三方存储，在flatMap()中完成查询，那么速度会非常之慢，查询需要等待返回后再将数据下发。期间如果数据量过大，很容易将第三方存储比如mysql服务打爆，因为每条数据都需要去查一遍mysql。所以这种方式非常不推荐。
1.9及之后
- 直接在DDL中定义数据维表，这里贴一下官网的定义方式：
CREATE TABLE MyUserTable (
...
) WITH (
'connector.type' = 'jdbc', -- required: specify this table type is jdbc
'connector.url' = 'jdbc:mysql://localhost:3306/flink-test', -- required: JDBC DB url
'connector.table' = 'jdbc_table_name', -- required: jdbc table name
'connector.driver' = 'com.mysql.jdbc.Driver', -- optional: the class name of the JDBC driver to use to connect to this URL.
-- If not set, it will automatically be derived from the URL.
'connector.username' = 'name', -- optional: jdbc user name and password
'connector.password' = 'password',
-- lookup options, optional, used in temporary join
'connector.lookup.cache.max-rows' = '5000', -- optional, max number of rows of lookup cache, over this value, the oldest rows will
-- be eliminated. "cache.max-rows" and "cache.ttl" options must all be specified if any
-- of them is specified. Cache is not enabled as default.
'connector.lookup.cache.ttl' = '10s', -- optional, the max time to live for each rows in lookup cache, over this time, the oldest rows
-- will be expired. "cache.max-rows" and "cache.ttl" options must all be specified if any of
-- them is specified. Cache is not enabled as default.
'connector.lookup.max-retries' = '3', -- optional, max retry times if lookup database failed
)
这里特别说明一下，connector.lookup.cache.max-rows 和 connector.lookup.cache.ttl必须同时出现。这里指的是配置缓存和缓存时间。
- 依旧存在几个小问题
- 指定缓存后，Join维表时拿到的可能是旧数据。所以，如果想要拿到尽可能新的数据，缓存时间又需要设置的较短或者没有，那又失去了缓存的意义；缓存条数如果过大会将内存撑爆，而且内存中的数据相对第三方存储中的数据来说，可能又是旧数据，而太小又得去频繁查询数据库，也就失去了缓存的意义。所以，缓存的大小和时间需要一个很好的平衡。不过通常来说，维表中的数据应该是很少变化的，同样，如果缓存较多的条数，那在启动任务时相应的提高TM的内存，也能够缓解撑爆内存的情形。
- 目前Flink Sql原生支持的维表关联只支持同步模式，如果需要异步模式或者想用其他的第三方存储只能够自己去实现。当然，我们之后也会有真正的异步维表Join案例分享，大家敬请期待。
接下来，我们通过案例来了解如何使用维表Join
Flink Sql 1.10 维表Join
在正式开始之前，先做好准备工作，之前通过docker安装的mysql存在一点小问题，我们往表中插入中文字符时会出现乱码，我们需要去解决一下。
首先，进入我们的mysql容器里面docker exec -it 9f7da4663dc1 '/bin/bash'，然后执行echo "character-set-server=utf8" >> /etc/mysql/mysql.conf.d/mysqld.cnf 。执行完毕后一路ctrl+D退到最外面，通过docker restart 容器id将mysql服务重启。重新连接上mysql，执行show variables like '%character%';，然后确定character_set_server是否为utf-8。
接着我们在mysql中创建一张维表
CREATE TABLE `dim_behavior` (
`id` int(11) NOT NULL AUTO_INCREMENT,
`c_name` varchar(255) COLLATE utf8_bin DEFAULT NULL,
`e_name` varchar(255) COLLATE utf8_bin DEFAULT NULL,
PRIMARY KEY (`id`)
)
再插入两条数据
INSERT INTO `test`.`dim_behavior`(`id`, `c_name`, `e_name`) VALUES (1, '浏览', 'pv');
INSERT INTO `test`.`dim_behavior`(`id`, `c_name`, `e_name`) VALUES (2, '购买', 'buy');
准备工作完毕，下面我们直接看代码
package FlinkSql;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.java.StreamTableEnvironment;
import org.apache.flink.types.Row;
import org.apache.flink.util.Collector;
public class FlinkSql03 {
public static final String CSV_TABLE_SOURCE_DDL = "" +
"CREATE TABLE csv_source (\n" +
" user_id bigint,\n" +
" item_id bigint,\n" +
" category_id bigint,\n" +
" behavior varchar,\n" +
" ts bigint,\n" +
" proctime as PROCTIME() \n"+
") WITH (\n" +
" 'connector.type' = 'filesystem', -- 指定连接类型\n" +
" 'connector.path' = 'C:\\Users\\tzmaj\\Desktop\\教程\\3\\UserBehavior.csv',-- 目录 \n" +
" 'format.type' = 'csv', -- 文件格式 \n" +
" 'format.field-delimiter' = ',' ,-- 字段分隔符 \n" +
" 'format.fields.0.name' = 'user_id',-- 第N字段名，相当于表的schema，索引从0开始 \n" +
" 'format.fields.0.data-type' = 'bigint',-- 字段类型\n" +
" 'format.fields.1.name' = 'item_id', \n" +
" 'format.fields.1.data-type' = 'bigint',\n" +
" 'format.fields.2.name' = 'category_id',\n" +
" 'format.fields.2.data-type' = 'bigint',\n" +
" 'format.fields.3.name' = 'behavior', \n" +
" 'format.fields.3.data-type' = 'String',\n" +
" 'format.fields.4.name' = 'ts', \n" +
" 'format.fields.4.data-type' = 'bigint'\n" +
") ";
public static final String MYSQL_TABLE_DIM_DDL = ""+
"CREATE TABLE `dim_behavior` (\n" +
" `id` int ,\n" +
" `c_name` varchar ,\n" +
" `e_name` varchar \n" +
")WITH (\n" +
" 'connector.type' = 'jdbc', -- 连接方式\n" +
" 'connector.url' = 'jdbc:mysql://localhost:3306/test', -- jdbc的url\n" +
" 'connector.table' = 'dim_behavior', -- 表名\n" +
" 'connector.driver' = 'com.mysql.jdbc.Driver', -- 驱动名字，可以不填，会自动从上面的jdbc url解析 \n" +
" 'connector.username' = 'root', -- 顾名思义 用户名\n" +
" 'connector.password' = '123456' , -- 密码\n" +
" 'connector.lookup.cache.max-rows' = '5000', -- 缓存条数 \n"+
" 'connector.lookup.cache.ttl' = '10s' -- 缓存时间 \n"+
")";
public static final String MYSQL_TABLE_SINK_DDL=""+
"CREATE TABLE `result_1` (\n" +
" `behavior` varchar ,\n" +
" `count_unique_user` bigint, \n" +
" `e_name` varchar \n" +
")WITH (\n" +
" 'connector.type' = 'jdbc', -- 连接方式\n" +
" 'connector.url' = 'jdbc:mysql://localhost:3306/test', -- jdbc的url\n" +
" 'connector.table' = 'result_1', -- 表名\n" +
" 'connector.driver' = 'com.mysql.jdbc.Driver', -- 驱动名字，可以不填，会自动从上面的jdbc url解析 \n" +
" 'connector.username' = 'root', -- 顾名思义 用户名\n" +
" 'connector.password' = '123456' , -- 密码\n" +
" 'connector.write.flush.max-rows' = '5000', -- 意思是攒满多少条才触发写入 \n" +
" 'connector.write.flush.interval' = '1' -- 意思是攒满多少秒才触发写入；这2个参数，无论数据满足哪个条件，就会触发写入\n" +
// " 'update-mode' = 'upsert' -- 指定为插入更新模式 \n"+
")";
public static void main(String[] args) throws Exception {
//构建StreamExecutionEnvironment
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
//构建EnvironmentSettings 并指定Blink Planner
EnvironmentSettings bsSettings = EnvironmentSettings.newInstance().useBlinkPlanner().inStreamingMode().build();
//构建StreamTableEnvironment
StreamTableEnvironment tEnv = StreamTableEnvironment.create(env, bsSettings);
//注册csv文件数据源表
tEnv.sqlUpdate(CSV_TABLE_SOURCE_DDL);
//注册mysql数据维表
tEnv.sqlUpdate(MYSQL_TABLE_DIM_DDL);
//注册mysql数据结果表
tEnv.sqlUpdate(MYSQL_TABLE_SINK_DDL);
//计算每种类型的行为有多少用户
Table group = tEnv.sqlQuery("select behavior,count(distinct user_id) count_unique_user from csv_source group by behavior ");
//转回datastream，因为需要增加proctime，而目前定义proctime方式只有两种，一种是在定义DDL的时候，一种是在DataStream转 Table的时候
//转撤回流是因为上面的sql用了group by，所以只能使用撤回流
DataStream<Row> ds = tEnv.toRetractStream(group, Row.class).flatMap(
new FlatMapFunction<Tuple2<Boolean, Row>, Row>() {
@Override
public void flatMap(Tuple2<Boolean, Row> value, Collector<Row> collect) throws Exception {
collect.collect(value.f1);
}
}
).returns(Types.ROW(Types.STRING,Types.LONG));
//给Table增加proctime字段，ts可以随便改成别的你喜欢的名字
Table table = tEnv.fromDataStream(ds, "behavior,count_unique_user,ts.proctime");
//建立视图，保留临时表
tEnv.createTemporaryView("group_by_view",table);
//pv，buy，cart...等行为对应的英文名，我们通过维表Join的方式，替换为中文名
//FOR SYSTEM_TIME AS OF a.ts AS b 这是固定写法，ts与上面指定dataStream schema时候用的名字一致
//这里之所以再group by，是让这次查询变成撤回流，这样插入mysql时，可以通过主键自动update数据
Table join = tEnv.sqlQuery("select b.c_name as behavior , max(a.count_unique_user) ,a.behavior as e_name " +
"from group_by_view a " +
"left join dim_behavior FOR SYSTEM_TIME AS OF a.ts AS b " +
"on a.behavior = b.e_name " +
"group by a.behavior,b.c_name");
//建立视图，保留临时表
tEnv.createTemporaryView("join_view",join);
//数据输出到mysql
tEnv.sqlUpdate("insert into result_1 select * from join_view");
//任务启动，这行必不可少！
env.execute("FlinkSql03");
}
}
如果UserBehavior.csv没有的话，可以通过第一课Flink Sql教程（1）最下方的连接下载。
接下来让我们看看mysql中的结果
behavior count_unique_user e_name
| behavior | count_unique_user | e_name |
|---|---|---|
| 购买 | 10587 | buy |
| 23143 | cart | |
| 12194 | fav | |
| 浏览 | 215662 | pv |
中间两行的behavior列之所以没有值，是因为我们没有在维表中定义，加上我们采用的是left join。大家可以自己定义然后测试数据能否正确补齐。
那么，我们这次的维表Join尝鲜就这样结束了，下一章将更新UDF和Temporal table function Join，不过应该是五一之后的事了。
