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

3、redis作为维表（带format）3.1、数据准备@Beforepublic void init() {RedisOptions.IS_TEST = true;RedisOperator redisOperator = RedisOperators.getSimple("10.201.0.33",6379,"123456",0);List<String> lists = new ArrayList<>();for (int i = 0; i < 1000; i++) {lists.add("{\n" +"\"number\": " + i + ",\n" +"\"name\": \"学生" + i + "\",\n" +"\"school\": \"学校" + ((i % 3) + 1) +"\",\n" +"\"class_id\": " + ((i % 10) + 1) +"\n" +"}");}/*** 初始化学生数据*/for (int i = 0; i < 1; i++) {redisOperator.rpush("students", lists.subList(1000 * i, 1000 * (i + 1)));}/*** 初始化班级数据*/for(int i = 0;i < 10;i++) {JSONObject jsonObject = new JSONObject();jsonObject.put("class_id",String.valueOf(i + 1));jsonObject.put("class_name","银河" + (i + 1) + "班");jsonObject.put("remark","remark" + i);redisOperator.set(String.valueOf(i + 1),jsonObject.toString());}/*** 初始化学校班级数据*/for(int j = 1;j < 4;j++) {for (int i = 1; i < 11; i++) {JSONObject jsonObject = new JSONObject();jsonObject.put("class_id",String.valueOf(i));jsonObject.put("class_name","银河" + i + "班");jsonObject.put("remark","remark" + i);jsonObject.put("school","学校" + j);redisOperator.hset("学校" + j, String.valueOf(i), jsonObject.toString());}}}3.2、使用GET作为维表查询命令@Testpublic void testGetSQL() throws Exception {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" +" )";/*** 这里测试的核心是维表有format=json配置项，有了format配置项后，字段个数不受限制，但是需要注意的是，作为get命令的key的字段* 一定要放在表申明的第一位，并且get命令的value的值使用format格式化后，比如是json格式，则json里一定要包含作为维表查询的*join on后面带的作为key的查询列，不然会报空指针异常*/String daeamon ="CREATE TABLE classes\n" +"(\n" +"class_idBIGINT,\n" +"class_namestring ,\n" +"remarkstring" +") \n" +"WITH (\n" +"'connector'='redis',\n" +"'host'='10.201.0.33', \n" +"'port'='6379',\n" +"'redis-mode'='single', \n" +"'format'='json', \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'='GET'\n" +" )";String sink ="CREATE TABLE sink_students\n" +"(\n" +"numberBIGINT ,\n" +"namestring,\n" +"schoolstring, \n" +"class_idBIGINT, \n" +"class_namestring, \n" +"remarkstring" +") \n" +"WITH (\n" +"'connector'='print'\n" +" )";tEnv.executeSql(source);tEnv.executeSql(daeamon);tEnv.executeSql(sink);String sql =" insert into sink_students "+ " select s.number,s.name,s.school,s.class_id,d.class_name,d.remarkfrom students s"+ "left join classes for system_time as of s.proctime as don d.class_id = s.class_id";TableResult tableResult = tEnv.executeSql(sql);tableResult.getJobClient().get().getJobExecutionResult().get();}3.3、使用HGET作为维表查询命令@Testpublic void testHGetSQL() throws Exception {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" +" )";/*** 这里测试的核心是维表有format=json配置项，有了format配置项后，字段个数不受限制，但是需要注意的是，作为hget命令的key的字段* 一定要放在表申明的第一位，field的字段一定要放在申明的第二位，并且hget命令的value的值使用format格式化后，比如是json格式，* 则json里一定要包含作为维表查询的 join on后面带的作为key和field的查询列，不然会报空指针异常*/String daeamon ="CREATE TABLE classes\n" +"(\n" +"schoolstring, \n" +"class_idBIGINT,\n" +"class_namestring," +"remarkstring" +") \n" +"WITH (\n" +"'connector'='redis',\n" +"'host'='10.201.0.33', \n" +"'port'='6379',\n" +"'format'='json', \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" +" )";String sink ="CREATE TABLE sink_students\n" +"(\n" +"numberBIGINT ,\n" +"namestring,\n" +"schoolstring, \n" +"class_idBIGINT, \n" +"class_namestring, \n" +"remarkstring" +") \n" +"WITH (\n" +"'connector'='print'\n" +" )";tEnv.executeSql(source);tEnv.executeSql(daeamon);tEnv.executeSql(sink);String sql =" insert into sink_students "+ " select s.number,s.name,s.school,s.class_id,d.class_name,d.remarkfrom students s"+ "left join classes for system_time as of s.proctime as don d.class_id = s.class_id and d.school = s.school";TableResult tableResult = tEnv.executeSql(sql);tableResult.getJobClient().get().getJobExecutionResult().get();}

flinksql读写redis( 三 )

经验总结扩展阅读

1、两个人在一起|总有一个人会改变自己

家用饭碗可以放烤箱吗

玉和翡翠假的怎么分辨

水培蔬菜有哪些,适合水培的蔬菜和水果

头条怎么才能涨粉

疯人愿原唱是谁疯了疯了伴奏

哪种面膜是你的菜

荔枝放冰箱能保存多久荔枝放冰箱能保存多长时间

男子结婚前夕|男子结婚前夕，妻子消失无踪，6年后逛街偶遇，他终于明白了

山楂糕不凝固能回锅重熬吗

会计学专业就业方向与就业前景怎么样

发型不要再披头散发了，脸上有这5个特征的女生，披头发真的很丑

多人运动事件是什么梗

五月一号怎么说早安五月一号早安句子精选

妙奇星球新手初阶怎么玩

美白的真谛是什么

因为42岁温峥嵘近照瘦脱相，皮包骨似大头娃娃，面对蛋糕一口不吃

秋天来了，女性不想肚子鼓鼓，推荐以下食物，排毒减肥，抗衰老！

智商|“以油养肤”是在交智商税？最全网红护肤油测评，拿走不谢！

墨鱼自蔽和什么寓言故事类似