基于Flink1.11.2 多表join与维表join

超、凢脫俗 2022-12-11 07:54 340阅读 0赞

1，业务场景：之前离线的业务场景，想通过实时来实现。离线业务是多个表join，因为接入的myslq binlog的数据，所以数据存在修改跟删除，这个是跟日志分析最大的区别，因为日志分析啥的数据只有新增，相对来说考虑的东西就少了很多。

2，首先我们要实现的是多表join，这个相对还是比较简单的，比如下面的SQL：

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMxODY2Nzkz_size_16_color_FFFFFF_t_70][]

3，但是3个表join可是全量join的，也就是说表的state是一直存在的，总会存在OOM爆炸的时候，那怎么办呢。

在我们使用window窗口的时候，一个窗口结束之后会帮我们清除state，这里Flink支持通过API清除state，不过是全局的，细粒度的还没研究。

因为业务是只管当天的数据，数据到kafka的时候会有延迟，但是不可能延迟一天，所以我这里设置TTL为1天到3天。

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMxODY2Nzkz_size_16_color_FFFFFF_t_70 1][]

当然，肯定要测试一下TTL，读取2个topic注册为表，然后join，30s之后发现join不上了。state过期了。这个大家手动去测试吧。

4，几个表join之后，再加上设置了 TTL，我们还要关联维表数据，在做之前我就想，维表会不会受TTL的影响，脑壳发昏了 。

经过鸡哥的提醒我才想起来了，维表读取的代码就是缓存，是guava。磕头谢罪。![20200929145049798.png][]

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMxODY2Nzkz_size_16_color_FFFFFF_t_70 2][]

5，维表的join，维表的join语法在官网一目了然：

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMxODY2Nzkz_size_16_color_FFFFFF_t_70 3][]

我们需要给主流添加上processTime:

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMxODY2Nzkz_size_16_color_FFFFFF_t_70 4][]

完整代码如下：

>     import java.util.concurrent.TimeUnit
>     
>     import org.apache.flink.api.common.restartstrategy.RestartStrategies
>     import org.apache.flink.api.common.time.Time
>     import org.apache.flink.contrib.streaming.state.RocksDBStateBackend
>     import org.apache.flink.runtime.state.StateBackend
>     import org.apache.flink.streaming.api.{CheckpointingMode, TimeCharacteristic}
>     import org.apache.flink.streaming.api.environment.CheckpointConfig
>     import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment
>     import org.apache.flink.table.api.EnvironmentSettings
>     import org.apache.flink.table.api.bridge.scala.StreamTableEnvironment
>     
>     /*
>       todo 在这里做join 操作 ttl 验证、
>      */
>     object practice_join_mysqlJoin {
>     
>       def main(args: Array[String]): Unit = {
>     
>         val env = StreamExecutionEnvironment.getExecutionEnvironment
>     
>         env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)
>         val bsSettings = EnvironmentSettings.newInstance.useBlinkPlanner.inStreamingMode.build
>         val stenv = StreamTableEnvironment.create(env, bsSettings)
>         val statement = stenv.createStatementSet()
>     
>         env.setRestartStrategy(RestartStrategies.fixedDelayRestart(5, Time.of(30L, TimeUnit.SECONDS)))
>         env.enableCheckpointing(1 * 60 * 1000L)
>         env.getCheckpointConfig.setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE)
>         env.getCheckpointConfig.setMinPauseBetweenCheckpoints(10000L)
>         env.getCheckpointConfig.setCheckpointTimeout(2 * 60000L)
>         env.getCheckpointConfig.setMaxConcurrentCheckpoints(1)
>         env.setStateBackend(new RocksDBStateBackend("hdfs://dev-ct6-dc-master01:8020/flink/checkpoints/rocksDBStateBackend2", true))
>         env.getCheckpointConfig.enableExternalizedCheckpoints(CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION)
>         //todo 设置状态过期时间
>     //    stenv.getConfig.setIdleStateRetentionTime(Time.days(1), Time.days(3))
>         stenv.getConfig.setIdleStateRetentionTime(Time.seconds(30), Time.minutes(6))
>     
>         //todo 1，从mysql同步数据到kafka，此略
>     
>         //todo 2,我们直接从dwd层获取数据,创建为table
>     
>         val assure_orders =
>           s"""
>              |CREATE TABLE assure_orders (
>              | order_id INT,
>              | order_name STRING,
>              | proctime as PROCTIME()
>              |) WITH (
>              | 'connector' = 'kafka',
>              | 'topic' = 'doris_test.assure_orders',
>              | 'properties.bootstrap.servers' = 'dev-ct6-dc-worker01:9092,dev-ct6-dc-worker02:9092,dev-ct6-dc-worker03:9092',
>              | 'properties.group.id' = 'test1',
>              | 'format' = 'canal-json',
>     //         | 'format' = 'json',
>              | 'scan.startup.mode' = 'earliest-offset'
>              |)
>            """.stripMargin
>         stenv.executeSql(assure_orders)
>     
>         val order_company =
>           s"""
>              |CREATE TABLE order_company (
>              | company_id INT,
>              | company_name STRING
>              |) WITH (
>              | 'connector' = 'kafka',
>              | 'topic' = 'doris_test.order_company',
>              | 'properties.bootstrap.servers' = 'dev-ct6-dc-worker01:9092,dev-ct6-dc-worker02:9092,dev-ct6-dc-worker03:9092',
>              | 'properties.group.id' = 'test1',
>              | 'format' = 'canal-json',
>     //         | 'format' = 'json',
>              | 'scan.startup.mode' = 'earliest-offset'
>              |)
>            """.stripMargin
>         stenv.executeSql(order_company)
>     
>     
>         val order_address =
>           s"""
>              |CREATE TABLE order_address (
>              | address_id INT,
>              | address_name STRING
>              |) WITH (
>              | 'connector' = 'kafka',
>              | 'topic' = 'doris_test.order_address',
>              | 'properties.bootstrap.servers' = 'dev-ct6-dc-worker01:9092,dev-ct6-dc-worker02:9092,dev-ct6-dc-worker03:9092',
>              | 'properties.group.id' = 'test1',
>              | 'format' = 'canal-json',
>     //         | 'format' = 'json',
>              | 'scan.startup.mode' = 'earliest-offset'
>              |)
>            """.stripMargin
>         stenv.executeSql(order_address)
>     
>         //todo ====================维表join====================
>         val dim_base_province_create =
>           """
>             | CREATE TABLE `dim_base_province` (
>             |      `id` INT,
>             |      `name` STRING,
>             |      `region_id` INT ,
>             |      `area_code`STRING,
>             |      PRIMARY KEY (id) NOT ENFORCED
>             |    ) WITH (
>             |     'connector' = 'jdbc',
>             |    'url' = 'jdbc:mysql://192.168.5.24:3306/doris_test?useUnicode=true&characterEncoding=UTF-8',
>             |    'table-name' = 'base_province_copy_copy', -- MySQL中的待插入数据的表
>             |      'driver' = 'com.mysql.jdbc.Driver',
>             |    'username' = 'root',
>             |    'password' = '123456',
>             |   'lookup.cache.max-rows'='1000',
>             |   'lookup.cache.ttl' = '5s'
>             |    )
>           """.stripMargin
>     
>         stenv.executeSql(dim_base_province_create)
>     
>         //todo  ====================打印测试=====================
>         val sink_print =
>           s"""
>              |create table sink_print (
>              |a INT,
>              |b STRING,
>              |c STRING,
>              |d STRING,
>              |e STRING
>              |) with ('connector' = 'print')
>             """.stripMargin
>     //    stenv.executeSql(sink_print)
>     
>         //todo 输出======================
>         val inset_kafka =
>           s"""
>              |CREATE TABLE sink_print (
>              |a INT,
>              |b STRING,
>              |c STRING,
>              |d STRING,
>              |e STRING
>              |) WITH (
>              | 'connector' = 'kafka',
>              | 'topic' = 'cccc',
>              | 'properties.bootstrap.servers' = 'dev-ct6-dc-worker01:9092,dev-ct6-dc-worker02:9092,dev-ct6-dc-worker03:9092',
>              | 'properties.group.id' = 'test1',
>              | 'format' = 'changelog-json',
>     //         | 'format' = 'json',
>              | 'scan.startup.mode' = 'earliest-offset'
>              |)
>            """.stripMargin
>     
>         stenv.executeSql(inset_kafka)
>     
>         val insert_sql  =
>           """
>             |INSERT INTO sink_print
>             | select d.order_id,d.company_name,d.address_name,d.order_name,dim_base_province.name from
>             | (
>             | SELECT a.*,b.*,c.*
>             | FROM assure_orders a
>             | JOIN
>             | order_company b
>             |  ON a.order_id = b.company_id
>             | JOIN
>             | order_address c
>             | ON a.order_id = c.address_id
>             |  ) d
>             | LEFT JOIN
>             | dim_base_province FOR SYSTEM_TIME AS OF d.proctime
>             | ON d.order_id = dim_base_province.id
>           """.stripMargin
>       /* val join_sql =
>          """
>            |INSERT INTO sink_print
>            |SELECT assure_orders.order_id,assure_orders.order_name,dim_base_province.name FROM assure_orders
>            |LEFT JOIN dim_base_province FOR SYSTEM_TIME AS OF assure_orders.proctime
>            |ON assure_orders.order_id = dim_base_province.id
>          """.stripMargin*/
>         stenv.executeSql(insert_sql)
>     
>       }
>     }

总结：

代码虽然看着简单，也算是简单的写了一些sql，后面也提交到测试集群运行了 ：

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMxODY2Nzkz_size_16_color_FFFFFF_t_70 5][]

后面的计划，我还是回归Streaming开发，SQL话开发虽然简单，但是对于调优跟底层的理解还是欠缺，回归streaming开发可能对我们的新人可能不太友好，但是对于深入理解，调优更有优势。

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMxODY2Nzkz_size_16_color_FFFFFF_t_70]: /images/20221123/f5b6540829f446f3a4af406ddcd2296d.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMxODY2Nzkz_size_16_color_FFFFFF_t_70 1]: /images/20221123/f6c49621086c40a185e2013c7132b874.png
[20200929145049798.png]: /images/20221123/3c018380cceb45419ee8a29cabd01286.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMxODY2Nzkz_size_16_color_FFFFFF_t_70 2]: /images/20221123/9a65302062154caf8f2b200e55801e02.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMxODY2Nzkz_size_16_color_FFFFFF_t_70 3]: /images/20221123/3c407a089e1f4160b79e9e4f57f446ba.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMxODY2Nzkz_size_16_color_FFFFFF_t_70 4]: /images/20221123/98756ca9adf6403e8bbbdda46c51c908.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMxODY2Nzkz_size_16_color_FFFFFF_t_70 5]: /images/20221123/20b264a4a2064845b36f02789579cc9b.png