数仓开发那些事（8）

程序员圣经
为什么刚刚能运行，现在就不行
为什么刚刚不运行，现在就可以
为什么他的可以跑，我的不能跑
为什么我的可以跑，他的就不行
为什么这台电脑能，那台就不行
为什么这台电脑不行，那台就行

神州员工：一闪，最近我有些感悟，和你分享一下👆

一闪：中国有句古话，叫做有句古话说得好.....

---------

老员工：一闪，来帮我看看这个问题，我Flink有个任务内存资源已经拉满了，为什么还是跑不动？

一闪：卧槽，15C60G，这么豪华的配置还跑不动？让我康康.....

CREATE TEMPORARY TABLE  xxx(tid              varchar      ,oid              varchar      ,num_iid          varchar     ,talent_flag      varchar     ,talent_id        varchar     ,talent_name      varchar     ,comb_goods_type 	BIGINT    
) WITH ('connector' = 'odps','endPoint' = 'xxx','project' = 'xxx','tableName' = 'xxx','accessId' = 'xxx'            ,'accessKey' = 'xxx','partition' = 'max_pt()','cache' = 'ALL','cacheSize' = '30000000'                             -- 缓存大小,'cacheTTLMs' = '3600000'                           -- 缓存重新加载的间隔时间); 

一闪：不对哇，你咋把一张事实表作为维表来用了，而且这个数据量有两千多万条？

老员工：没办法啊，我现在要的维度数据只在这张事实表里面才有，所以我只能用它做维表。引用odps（阿里存储，类似于hdfs）的表作为维表的时候，一定要全量缓存一次（即cacheSize参数一定要大于表内数据条数），所以我就这样配置了.....

一闪：这个维表一定会广播到所有的并发上，所以这个维表会占用大量内存，目前看下来就是这个大维表的问题，思路就是不能让全量维表分发到每个并发上。直接掏出我的秘籍（浏览器输入flink.apache.org）

----经过一番搜索，在官网上找到了如下信息：

Hints | Apache Flink

SHUFFLE_HASH suggests that Flink uses Shuffle Hash join. The join side with the hint will be the join build side, it performs well when the data volume of the hint side of table is not too large.

Note: SHUFFLE_HASH only supports join with equivalence join condition.

--Example↓
CREATE TABLE t1 (id BIGINT, name STRING, age INT) WITH (...);
CREATE TABLE t2 (id BIGINT, name STRING, age INT) WITH (...);
CREATE TABLE t3 (id BIGINT, name STRING, age INT) WITH (...);-- Flink will use hash join and t1 will be the build side.
SELECT /*+ SHUFFLE_HASH(t1) */ * FROM t1 JOIN t2 ON t1.id = t2.id;-- Flink will use hash join for both joins and t1, t3 will be the join build side.
SELECT /*+ SHUFFLE_HASH(t1, t3) */ * FROM t1 JOIN t2 ON t1.id = t2.id JOIN t3 ON t1.id = t3.id;-- SHUFFLE_HASH don't support non-equivalent join conditions.
-- For this case, Join Hint will not work, and only nested loop join can be applied.
SELECT /*+ SHUFFLE_HASH(t1) */ * FROM t1 join t2 ON t1.id > t2.id;

-----

一闪：所以我们只要把这张大维表加一下shuffle_hash的join hint就可以了，直接就完美了。堆内存就可以省下来一部分。

老员工：（卧槽，还给我扯内存模型，有点装X）那我问你，理论上Flink中TM托管内存的大小应该是TM内存的40%，那么为什么我从FlinkUI上看，实际上分配给托管内存的大小不到40%？

一闪：因为Flink会保留一部分内存用于JVM运行时、本地代码和一些基础操作系统功能。这部分内存不会用于Flink的直接作业内存分配。所以6.7GB并不是简单通过TM总内存乘40%直接算出来的。

老员工：这年轻人，卧槽