Flink SQL realizes reading and writing redis and dynamically generates hset key

        

         In the official offering flink-connect-redis It doesn't work Flink SQL Read and write in a different way redis. If you want to achieve Flink sql Read and write in a different way redis You need to implement the corresponding code yourself , Fortunately GitHub There is already a corresponding open source implementation , We only need to use the corresponding reference .

pom rely on ：

<dependency>
    <groupId>io.github.jeff-zou</groupId>
    <artifactId>flink-connector-redis</artifactId>
    <version>1.0.11</version>
</dependency>

By importing jar, Can achieve Flink sql Reading and writing redis ：

First create Flink CDC A watch of ：

CREATE TABLE memberCDC(
  `member_id`                  bigint,
   mobile string,
  `tenant_id`           string   ,
  gender                int ,
   proctime as procTime(),
     PRIMARY KEY(member_id) NOT ENFORCED
     ) WITH (
     'connector' = 'mysql-cdc',
     'hostname' = '10.50.20.181',
     'port' = '330806',
     'username' = 'reGaYTkly',
     'password' = '40IGOJHFDlloBE6',
     'database-name' = 'member_db',
     'table-name' = 'm_member_tenant',
   'scan.startup.mode'='latest-offset',
   'debezium.skipped.operations'='d'
   )

Flink SQL write in redis（ With Hset For example, data type ）：

CREATE TABLE memberRedis (
`member_id` bigint,
mobile string,
`tenant_id` string,
gender int
) WITH (
'connector' = 'redis',
'host' = '10.180.80.89',
'port' = '300079',
'password' = 'big88dajjggta34',
'redis-mode' = 'single',
'command'='hset'
)

inserter into memberRedis
select
member_id,
mobile,
tenant_id,
gender
from memberCDC

     here member_id amount to Hset Of   additionalKey mobile amount to key, And this can be done through SQL The dynamic splicing of statements is needed key Format . It is equivalent to supporting dynamic generation of what you need key.

Flink SQL Read redis（ With Hset For example, data type ）：

Read redis Data in , It is generally used as a dimension Association ：

CREATE TABLE memberSourceRedis (
`member_id` bigint,
mobile string,
`tenant_id` string,
gender int
) WITH (
'connector' = 'redis',
'host' = '10.110.60.5',
'port' = '30479',
'password' = 'daffdag1234',
'redis-mode' = 'single',
'command'='hget',
'maxIdle'='2',
'minIdle'='1',
'lookup.cache.max-rows'='10',
'lookup.cache.ttl'='10',
'lookup.max-retries'='3'
)

insert into memberRedis
select
t2.member_id,
t2.mobile,
t2.tenant_id,
t1.gender
from memberCDC t1
left join memberSourceRedis for system_time as of t1.proctime as t2 on
t1.member_id = t2.member_id and t1.mobile = t2.mobile

     Similar reading , here member_id amount to Hset Of   additionalKey mobile amount to key, Here you can go through SQL Statement dynamic splicing generation key, incoming key It's actually memberCDC The data value of the associated field in .

Connect redis parameter list ：

summary ：

     Above Flink SQL Reading and writing redis about hset Data types support writing and reading values of a field , Multiple field writes will be filtered by default .

     If you want to support writing and reading of multiple fields , The dependent code needs to be modified , When writing, multiple fields are encapsulated as JSONString write in Redis, When reading, the JSON Data is passed through catalog Map to values corresponding to multiple fields .

    

---

     Subsequent articles will also introduce how to modify the source code , compile jar Package to support multiple fields Flink SQL Read and write redis. If you need it, you can also use the following link to realize the corresponding flink sql Reading and writing Redis Special business needs . Feel useful and welcome to pay attention     Dida     official account ~~

     In this paper, the reference GitHub Open source projects for ：https://github.com/jeff-zou/flink-connector-redis