图文精华

About云-梭伦科技»专题 › 技术学习(版主发帖区) › 大数据学习 › Flink › 技术应用› 实战：Flink 1.12 维表 Join Hive 最新分区功能实现
分享

实战：Flink 1.12 维表 Join Hive 最新分区功能实现

levycui 2020-12-23 19:54:34 发表于 技术应用 [显示全部楼层] 只看大图 回帖奖励 阅读模式 关闭右栏 0 3152
About云VIP帮助找工作和提升
本帖最后由 levycui 于 2020-12-23 20:09 编辑
问题导读：
1、如何设计Flink 1.12 前关联 Hive 最新分区方案？
2、如何设计Flink 1.12 关联 Hive 最新分区表？
3、如何实现关联 Hive 最新分区 Demo？
4、如何实现Flink 事实表与 Hive 最新分区数据关联？


我们生产常有将实时数据流与 Hive 维表 join 来丰富数据的需求，其中 Hive 表是分区表，业务上需要关联上 Hive 最新分区的数据。上周 Flink 1.12 发布了，刚好支撑了这种业务场景，我也将 1.12 版本部署后做了一个线上需求并上线。对比之前生产环境中实现方案，最新分区直接作为时态表提升了很多开发效率，在这里做一些小的分享。

  •     Flink 1.12 前关联 Hive 最新分区方案
  •     Flink 1.12 关联 Hive 最新分区表
  •     关联Hive最新分区 Demo
  •     Flink SQL 开发小技巧

Flink 1.12 前关联 Hive 最新分区方案

在分区时态表出来之前，为了定期关联出最新的分区数据，通常要写 DataStream 程序，在 map 算子中实现关联 Hive 最新分区表的逻辑，得到关联打宽后的 DataStream 对象，通过将该 DataStream 对象转换成 Table 对象后，再进行后续的 SQL 业务逻辑加工。

  1. StreamTableEnvironment tblEnv = StreamTableEnvironment.create(env, streamSettings);  

  2. DataStream<Tuple2<MasterBean, HiveDayIndexBean>> indexBeanStream = masterDataStream.map(new IndexOrderJoin());  

复制代码

map 算子中的主要逻辑: 将 T+2 的维度数据与实时数据关联，返回 Tuple2<MasterBean, HiveDimBean> 数据，因为离线数仓出数一般在凌晨 3 点，有时候由于集群资源不稳定导致数据产出慢，业务对实时性要求也不高，所以这里用的是 T+2 的数据。

  1. public class IndexOrderJoin extends RichMapFunction<MasterBean, Tuple2<MasterBean, HiveDimBean>> {    

  2.     private Map<Integer, Map<String, HiveDimBean>> map = null;  

  3.     Logger logger;  

  4.   

  5.     @Override  

  6.     public void open(Configuration parameters) throws Exception {  

  7.         logger = LoggerFactory.getLogger(Class.forName("com.hll.util.IndexOrderJoin"));  

  8.         map = new HashMap<>(); 

  9.     }  

  10.   

  11. public Tuple2<MasterBean, HiveDayIndexBean> map(MasterBean masterBean) {    

  12.     if (map.get(masterBean.getReportDate() - 2) == null) {    

  13.         //如果map里没有T+2的维表数据则查询一次Hive，并将结果存入线程级别map，所以保证Task维表数据是全的    

  14.         logger.info("initial hive data : {}", masterBean.getReportDate());    

  15.         map.put(masterBean.getReportDate() - 2, getHiveDayIndex(masterBean.getReportDate() - 2));    

  16.     }    

  17.     //将的kafka数据与hive join后返回打宽数据    

  18.     return new Tuple2<>(masterBean, map.get(masterBean.getReportDate() - 2).get(masterBean.getGroupID()));    

  19. }

复制代码

基于关联打宽后的 DataStream 创建视图，然后再做后续的 SQL 业务逻辑查询。
  1. 

  2. tblEnv.createTemporaryView("index_order_master", indexBeanStream); tblEnv.sqlUpdate("select group_id, group_name, sum(amt) from index_order_master  group by group_id, group_name"); 

  3. tblEnv.execute("rt_aggr_master_flink");

复制代码

可以看出，在没有支持 Hive 最新分区做时态表的时候，简单的一个 join 便涉及到DataStream、map 算子，程序的代码量和维护成本会是比较大的。


Flink 1.12 关联 Hive 最新分区表

Flink 1.12 支持了 Hive 最新的分区作为时态表的功能，可以通过 SQL 的方式直接关联 Hive 分区表的最新分区，并且会自动监听最新的 Hive 分区，当监控到新的分区后，会自动地做维表数据的全量替换。通过这种方式，用户无需编写 DataStream 程序即可完成 Kafka 流实时关联最新的 Hive 分区实现数据打宽。

2020-12-23_194922.jpg

图片出自徐榜江（雪尽）在 FFA 2020 上的分享


  •     参数解释

■ streaming-source.enable 开启流式读取 Hive 数据。
■ streaming-source.partition.include
1.latest 属性: 只读取最新分区数据。
2.all: 读取全量分区数据 ，默认值为 all，表示读所有分区，latest 只能用在 temporal join 中，用于读取最新分区作为维表，不能直接读取最新分区数据。

■ streaming-source.monitor-interval 监听新分区生成的时间、不宜过短 、最短是1 个小时，因为目前的实现是每个 task 都会查询 metastore，高频的查可能会对metastore 产生过大的压力。需要注意的是，1.12.1 放开了这个限制，但仍建议按照实际业务不要配个太短的 interval。

■ streaming-source.partition-order 分区策略

主要有以下 3 种，其中最为推荐的是 partition-name：

1.partition-name 使用默认分区名称顺序加载最新分区
2.create-time 使用分区文件创建时间顺序
3.partition-time 使用分区时间顺序

  •     具体配置

使用 Hive 最新分区作为 Tempmoral table 之前，需要设置必要的两个参数：
  1. 'streaming-source.enable' = 'true',  

  2. 'streaming-source.partition.include' = 'latest'
复制代码

我们可以再创建一张基于 Hive 表的新表，在 DDL 的 properties 里指定这两个参数，也可以使用 SQL Hint 功能，在使用时通过 SQL Hint 指定 query 中表的参数。以使用 SQL Hint 为例，我们需要用 /* option */  指定表的属性参数，例如：

  1. SELECT * FROM hive_table /*+ OPTIONS('streaming-source.enable'='true',

  2. 'streaming-source.partition.include' = 'latest') */;  
复制代码

我们需要显示地开启 SQL Hint 功能, 在 SQL Client 中可以用 set 命令设置：

  1. set table.dynamic-table-options.enabled= true;
复制代码

在程序代码中，可以通过 TableConfig 配置：

  1. tblEnv.getConfig().getConfiguration().setString("table.dynamic-table-options.enabled", 

  2. "true");  

复制代码

Flink 官网也给出了一个详细的例子，这里也简单说明下。
  1. --将方言设置为hive以使用hive语法  

  2. SET table.sql-dialect=hive;      

  3. CREATE TABLE dimension_table (      

  4.   product_id STRING,      

  5.   product_name STRING,      

  6.   unit_price DECIMAL(10, 4),      

  7.   pv_count BIGINT,      

  8.   like_count BIGINT,      

  9.   comment_count BIGINT,      

  10.   update_time TIMESTAMP(3),      

  11.   update_user STRING,      

  12.   ...      

  13. ) PARTITIONED BY (pt_year STRING, pt_month STRING, pt_day STRING) TBLPROPERTIES (      

  14.   -- 在创建hive时态表时指定属性      

  15.   'streaming-source.enable' = 'true',      

  16.   'streaming-source.partition.include' = 'latest',      

  17.   'streaming-source.monitor-interval' = '12 h',      

  18.   'streaming-source.partition-order' = 'partition-name',  -- 监听partition-name最新分区数据    

  19. );      
复制代码
  1. --将方言设置为default以使用flink语法  

  2. SET table.sql-dialect=default;      

  3. CREATE TABLE orders_table (      

  4.   order_id STRING,      

  5.   order_amount DOUBLE,      

  6.   product_id STRING,      

  7.   log_ts TIMESTAMP(3),      

  8.   proctime as PROCTIME()      

  9. ) WITH (...);      

复制代码
  1. --将流表与hive最新分区数据关联  

  2. SELECT * FROM orders_table AS order     

  3. JOIN dimension_table FOR SYSTEM_TIME AS OF o.proctime AS dim    

  4. ON order.product_id = dim.product_id;
复制代码

关联 Hive 最新分区 Demo

  •     工程依赖

将 Demo 工程中使用到的 connector 和 format 依赖贴到这里，方便大家本地测试时参考。


  1. <dependencies>  

  2.     

  3.   

  4.     <dependency>  

  5.         <groupId>mysql</groupId>  

  6.         <artifactId>mysql-connector-java</artifactId>  

  7.         <version>${mysql.version}</version>  

  8.         <!--<scope>provided</scope>-->  

  9.     </dependency>  

  10.   

  11.     <dependency>  

  12.         <groupId>org.apache.flink</groupId>  

  13.         <artifactId>flink-connector-jdbc_2.12</artifactId>  

  14.         <version>${flink.version}</version>  

  15.     </dependency>  

  16.  

  17.     <dependency>  

  18.         <groupId>org.apache.flink</groupId>  

  19.         <artifactId>flink-sql-connector-kafka_2.11</artifactId>  

  20.         <version>${flink.version}</version>  

  21.         <scope>provided</scope>  

  22.     </dependency>  

  23.   

  24.     <dependency>  

  25.         <groupId>org.apache.flink</groupId>  

  26.         <artifactId>flink-json</artifactId>  

  27.         <version>${flink.version}</version>  

  28.         <scope>provided</scope>  

  29.     </dependency>  

  30.   

  31.     <dependency>  

  32.         <groupId>org.apache.flink</groupId>  

  33.         <artifactId>flink-connector-hive_${scala.binary.version}</artifactId>  

  34.         <version>${flink.version}</version>  

  35.         <scope>provided</scope>  

  36.     </dependency>  

  37.   

  38.     <dependency>  

  39.         <groupId>org.apache.hive</groupId>  

  40.         <artifactId>hive-exec</artifactId>  

  41.         <version>3.1.0</version>  

  42.     </dependency>  

  43.   

  44. </dependencies>  
复制代码

  •     在 Sql Client 中注册 HiveCatalog：

  1. vim conf/sql-client-defaults.yaml 

  2. catalogs: 

  3.   - name: hive_catalog 

  4.     type: hive 

  5.     hive-conf-dir: /disk0/soft/hive-conf/ #该目录需要包hive-site.xml文件  
复制代码

  •     创建 Kafka 表
  1. CREATE TABLE hive_catalog.flink_db.kfk_fact_bill_master_12 (  

  2.     master Row<reportDate String, groupID int, shopID int, shopName String, action int, orderStatus int, orderKey String, actionTime bigint, areaName String, paidAmount double, foodAmount double, startTime String, person double, orderSubType int, checkoutTime String>,  

  3. proctime as PROCTIME()  -- PROCTIME用来和Hive时态表关联  

  4. ) WITH (  

  5.  'connector' = 'kafka',  

  6.  'topic' = 'topic_name',  

  7.  'format' = 'json',  

  8.  'properties.bootstrap.servers' = 'host:9092',  

  9.  'properties.group.id' = 'flinkTestGroup',  

  10.  'scan.startup.mode' = 'timestamp',  

  11.  'scan.startup.timestamp-millis' = '1607844694000'  

  12. );
复制代码


  •     Flink 事实表与 Hive 最新分区数据关联

dim_extend_shop_info 是 Hive 中已存在的表，所以我们下面用 table hint 动态地开启维表参数。

  1. CREATE VIEW IF NOT EXISTS hive_catalog.flink_db.view_fact_bill_master as  

  2. SELECT * FROM  

  3.  (select t1.*, t2.group_id, t2.shop_id, t2.group_name, t2.shop_name, t2.brand_id,   

  4.      ROW_NUMBER() OVER (PARTITION BY groupID, shopID, orderKey ORDER BY actionTime desc) rn  

  5.     from hive_catalog.flink_db.kfk_fact_bill_master_12 t1  

  6.        JOIN hive_catalog.flink_db.dim_extend_shop_info   

  7.      /*+ OPTIONS('streaming-source.enable'='true',             'streaming-source.partition.include' = 'latest',  

  8.     'streaming-source.monitor-interval' = '1 h',

  9.     'streaming-source.partition-order' = 'partition-name') */ FOR SYSTEM_TIME AS OF t1.proctime AS t2 --时态表  

  10. ON t1.groupID = t2.group_id and t1.shopID = t2.shop_id  

  11.     where groupID in (202042)) t  where t.rn = 1  
复制代码

  •     结果数据 Sink 到 MySQL

  1. CREATE TABLE hive_catalog.flink_db_sink.rt_aggr_bill_food_unit_rollup_flk (  

  2.       report_date String,  

  3.       group_id int,  

  4.       group_name String,  

  5.       shop_id int,  

  6.       shop_name String,  

  7.       brand_id BIGINT,  

  8.       brand_name String,  

  9.       province_name String,  

  10.       city_name String,  

  11.       foodcategory_name String,  

  12.       food_name String,  

  13.       food_code String,  

  14.       unit String,  

  15.       rt_food_unit_cnt double,  

  16.       rt_food_unit_amt double,  

  17.       rt_food_unit_real_amt double,  

  18.     PRIMARY KEY (report_date, group_id, shop_id, brand_id, foodcategory_name, food_name, food_code, unit) NOT ENFORCED) WITH (  

  19.     'connector' = 'jdbc',   

  20.     'url' = 'jdbc:mysql://host:4400/db_name?autoReconnect=true&useSSL=false',  

  21.     'table-name' = 'table-name',   

  22.     'username' = 'username',   

  23.     'password' = 'password'  

  24. )  
复制代码
  1. insert into hive_catalog.flink_db_sink.rt_aggr_bill_food_unit_rollup_flk  

  2. select reportDate, group_id, group_name, shop_id, shop_name, brand_id, brand_name, province_name, city_name 

  3.    , SUM(foodNumber)  rt_food_cnt  

  4.    , sum(paidAmount)  rt_food_amt  

  5.    , sum(foodAmount)  rt_food_real_amt  

  6.    from  hive_catalog.flink_db.view_fact_bill_master  

  7.    group by reportDate, group_id, group_name, shop_id, shop_name, brand_id, brand_name, province_name, city_name;
复制代码

  •     ORC format 的 BUG

在读取 ORC format 的表时，无法读取数据，我也向社区提了一个 Jira: https://issues.apache.org/jira/browse/FLINK-20576，读取其他 format 的表不存在问题，本地测试了读取 parquet 和 csv 都是正常的。

总结下上面的代码，只需通过 Flink SQL 便能实现 Kafka 实时数据流关联最新的 Hive 分区。同时我们结合了 HiveCatalog，可以复用 hive 的表和已经创建过的 kafka source 表，MySql sink 表，使得程序只需要关心具体的业务逻辑，无需关注 source/sink 表的创建，提高了代码的复用性以及可读性。对比之前的方案，纯 SQL 的开发显然降低了开发维护成本和用户门槛。

Flink SQL 开发小技巧
  •     结合 Hive catalog，持久化 source 与 sink 表，减少重复建表，使得代码只需关注逻辑 SQL。
  •     结合 Flink 视图，组织好业务加工逻辑，提高 SQL 的可读性。
  •     利用 SQL Client 调试 SQL，程序没问题后再打包上线，而不是直接提交到集群做测试。


作者：余东@哗啦啦
来源：https://mp.weixin.qq.com/s/yh4StVZ68aqfHFlMH_Sadw

最新经典文章，欢迎关注公众号
回复

使用道具 举报

自定义广告语

没找到任何评论，期待你打破沉寂

高级模式
B Color Image Link Quote Code Smilies
您需要登录后才可以回帖 登录 | 立即注册

本版积分规则

发表新帖
levycui

超级版主

关注

  • 650

    主题

  • 1163

    帖子

  • 97

    粉丝

TA的主题
关闭

推荐上一条 /2 下一条

关于我们· 联系我们· 加入我们· 小黑屋· 合作伙伴   

Copyright © 2001-2024 About云-梭伦科技 Powered by Discuz! X3.4 Licensed Discuz Team.

简书 / Smrz 京ICP备2020039040号 Wxb 简书网举报电话：021-34700000