blinksql 通过维表关联实现hash结构的数据管理

矫情吗；* 2022-10-05 00:57 119阅读 0赞

1、相关sql如下：

CREATE TABLE source_table (
      --必须和Kafka源表中的5个字段的顺序保持一致。
      messageKey VARBINARY,
      `message` VARBINARY,
      `topic` VARCHAR,
      `partition` INT,
      `offset` BIGINT
    ) WITH (
      `type` = 'kafka011',
      `topic` = 'flink005',
      `group.id` = 'kafka_test_wpp',
      `bootstrap.servers` = 'xxxx:9092,xxxx:9092,xxxx:9092'
     -- `startupMode` = 'TIMESTAMP'
    );
    
    
    create view source_table1 as 
    SELECT 
    cast(`message` as VARCHAR ) msg 
    from source_table;
    
    
    create table redis_dim (
        id VARCHAR,
        val VARCHAR,
        PRIMARY KEY (id), 
        PERIOD FOR SYSTEM_TIME
      ) with (
      type = 'redis',
      host = 'xxxx.redis.rds.aliyuncs.com',
      port = '6379',
      dbNum = '0',
      password = 'xxxx',
      hashName='hkeyvalue'
    );
    
    
    create table sink_print2(
        msg varchar,
        redisval VARCHAR
    )with(
        type='print'
    );
    
    
    insert into sink_print2 
    SELECT 
    e.msg, 
    cast(w.val as VARCHAR ) as redisval
    from source_table1 as e left join redis_dim FOR SYSTEM_TIME AS OF PROCTIME() AS w
    on e.msg = w.id
    ;

2、kafka的数据如下：

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dwcHdwcDE_size_16_color_FFFFFF_t_70][]

3、redis的维表数据

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dwcHdwcDE_size_16_color_FFFFFF_t_70 1][]

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dwcHdwcDE_size_16_color_FFFFFF_t_70]: /images/20221005/e89d8c5f6f5b4b0eb50222487219b90a.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dwcHdwcDE_size_16_color_FFFFFF_t_70 1]: /images/20221005/3790764c3c5d48e38eb21e640620ff1c.png