flinksql读写redis( 五 ) _生活百科

4.4、使用HSET命令作为sink表写入命令（指定key）@Testpublic void testHSetWithKey() throws Exception {long start = System.currentTimeMillis();StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();EnvironmentSettings environmentSettings =EnvironmentSettings.newInstance().useBlinkPlanner().inStreamingMode().build();StreamTableEnvironment tEnv = StreamTableEnvironment.create(env, environmentSettings);String source ="CREATE TABLE students\n" +"(\n" +"numberBIGINT ,\n" +"namestring,\n" +"schoolstring, \n" +"class_idBIGINT, \n" +"proctime as PROCTIME() \n" +") \n" +"WITH (\n" +"'connector'='redis',\n" +"'host'='10.201.0.33', \n" +"'port'='6379',\n" +"'redis-mode'='single', \n" +"'password'='123456',\n" +"'database'='0',\n" +"'key'='students',\n" +"'format'='json',\n" +"'batch-fetch-rows'='1000',\n" +"'json.fail-on-missing-field' = 'false',\n" +"'json.ignore-parse-errors' = 'true',\n" +"'command'='BLPOP'\n" +" )";String daeamon ="CREATE TABLE classes\n" +"(\n" +"schoolstring, \n" +"class_idBIGINT,\n" +"class_namestring" +") \n" +"WITH (\n" +"'connector'='redis',\n" +"'host'='10.201.0.33', \n" +"'port'='6379',\n" +"'redis-mode'='single', \n" +"'password'='123456',\n" +"'lookup.cache.max-rows'='1000',\n" +"'lookup.cache.ttl'='3600',\n" +"'lookup.cache.load-all'='true',\n" +"'database'='0',\n" +"'command'='HGET'\n" +" )";/***1、这里因为command是HSET，所以需要一个key和一个field，这里配置项指定了key，那么主键拼接就作为field，*使用hset保存到redis*2、作为sink有一个sink.key.ttl参数可以设置key保存在redis的ttl生存时间，单位秒，默认-1表示长期保存*/String sink ="CREATE TABLE sink_students\n" +"(\n" +"schoolstring, \n" +"numberBIGINT ,\n" +"namestring,\n" +"class_idBIGINT, \n" +"class_namestring, \n" +"primary key(number) not enforced" +") \n" +"WITH (\n" +"'connector'='redis',\n" +"'host'='10.201.0.33', \n" +"'port'='6379',\n" +"'redis-mode'='single', \n" +"'password'='123456',\n" +"'database'='0',\n" +"'format'='json',\n" +"'key'='sink_students_hset',\n" +"'batch-fetch-rows'='1000',\n" +"'json.fail-on-missing-field' = 'false',\n" +"'json.ignore-parse-errors' = 'true',\n" +"'sink.parallelism' = '16',\n" +"'sink.key.ttl' = '300',\n" +"'command'='HSET'\n" +" )";tEnv.executeSql(source);tEnv.executeSql(daeamon);tEnv.executeSql(sink);String sql =" insert into sink_students "+ " select s.school,s.number,s.name,s.class_id,d.class_namefrom students s"+ " left join classes for system_time as of s.proctime as d on d.class_id = s.class_id and d.school = s.school";TableResult tableResult = tEnv.executeSql(sql);tableResult.getJobClient().get().getJobExecutionResult().get();long end = System.currentTimeMillis();System.out.println("耗时：" + (end - start) + "ms");}5、配置说明配置项描述hostredis的hostportredis的portpasswordredis的passwordcluster-nodesredis的集群节点,ip和端口之间用英文冒号分隔，多个ip端口用英文逗号分割master.nameredis的sentinel模式的master节点的名称sentinels.inforedis的sentinel模式的info信息sentinels.passwordredis的sentinel模式的密码databaseredis的database，一般是0~15commandredis的命令，作为流表时支持BLPOP、BRPOP、LPOP、RPOP、SPOP；作为维表时支持GET、HGET；作为sink表时支持LPUSH、RPUSH、SADD、SET、HSETredis-moderedis的部署模式，single、cluster、sentinelkeyredis需要访问的key ，比如数据是以某个固定的key存放在redis里，值是一个list；redis执行lpush、rpush、sadd、hset等sink使用的命令时的key；timeout连接redis的超时时间，单位毫秒max-total连接redis的连接池的最大连接数max-idle连接redis的连接池的最大空闲数min-idle连接redis的连接池的最小空闲数format格式化数据格式，如json、csvbatch-fetch-rows像LPOP、BLPOP、RPOP、BRPOP这种命令每次从redis拿到数据的条数lookup.cache.max-rows作为维表lookup模式，缓存在内存中的数据的最大条数lookup.cache.ttl作为维表lookup模式，缓存在内存中的数据的ttl超时时间，单位秒lookup.max-retries作为维表lookup模式，查找数据的失败重试次数lookup.cache.load-all作为维表lookup模式，查找数据是否加载所有，主要是针对hget命令，如：HGET KEY_NAME FIELD_NAME；是否根据key查出所有field的值，这里可以根据实际hash表的大小决定是否要查询所有出来缓存起来sink.max-retriesredis作为sink源时，最大重试次数sink.parallelismredis作为sink源时，sink的并行数，默认并行度为核心数sink.key.ttlredis作为sink源时，sink的数据保存在redis的ttl超时时间，单位秒，默认为-1表示长期保存lookup.max-retries作为维表lookup模式，查找数据的失败重试次数源码地址：https://gitee.com/rongdi/flinksql-connector-redis/
【flinksql读写redis】

flinksql读写redis( 五 )

推荐阅读

带梓的女孩名字及寓意

学生会活动策划方案范文经典学生会活动策划范文

12星座女谁是小野猫谁是小奶猫

月光石手链有什么寓意月光石有什么寓意

粉色的粘土小玫瑰diy

倚的组词倚的组词组

2023武汉中考志愿填报规则详解 2021年武汉中考志愿填报时间

关于深圳市万科教育发展基金会简述深圳市万科教育发展基金会

没有足够的可用内存来运行此程序没有足够的可用内存来运行程序

缓刑是什么意思还要坐牢吗缓刑是什么意思

有一个“撩妹高手”的儿子，是种什么样的体验？

南阳多少人「南阳多少人口2020总人数」

什么是事故车,如何鉴定车子撞到什么程度算事故车

秋季喝什么汤最好秋季喝什么汤最养生

绝地求生刺激战场军团名字怎么改？军团名字修改攻略[多图]

梦见大鱼吃小鱼

文房四宝指什么文房四宝指什么数字

OPPO Reno6简单评测 oppo reno6有nfc吗

怎么判断二手车的真实行驶里程? 要看哪些细节

双子座成功人士知多少？