Flink操练（六十）之数据写入到redis

　　关键代码 public static class MyRedisMapper implements  RedisMapper>     {         //获取key         @Override         public RedisCommandDescription getCommandDescription() {             return new RedisCommandDescription(RedisCommand.HSET,＂tuple＂);         }         //获取key的数据         @Override         public String getKeyFromData(Tuple2 in) {             return in.f0;         }         //获取filed的数据         @Override         public String getValueFromData(Tuple2 in) {             return in.f1.toString();         }     }完整代码package day07; import org.apache.flink.api.java.tuple.Tuple2; import org.apache.flink.streaming.api.datastream.DataStreamSource; import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment; import org.apache.flink.streaming.connectors.redis.RedisSink; import org.apache.flink.streaming.connectors.redis.common.config.FlinkJedisPoolConfig; import org.apache.flink.streaming.connectors.redis.common.mapper.RedisCommand; import org.apache.flink.streaming.connectors.redis.common.mapper.RedisCommandDescription; import org.apache.flink.streaming.connectors.redis.common.mapper.RedisMapper; /**  * @program: bigData_learn  * @description: 写入到Redis  * @author: Mr.逗  * @create: 2021-09-26 10:24  **/ public class WriteToRedis {     public static class MyRedisMapper implements  RedisMapper>     {         //获取key         @Override         public RedisCommandDescription getCommandDescription() {             return new RedisCommandDescription(RedisCommand.HSET,＂tuple＂);         }         //获取key的数据         @Override         public String getKeyFromData(Tuple2 in) {             return in.f0;         }         //获取filed的数据         @Override         public String getValueFromData(Tuple2 in) {             return in.f1.toString();         }     }     public static void main(String[] args) {         StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();         env.setParallelism(1);         DataStreamSource> stream = env                 .fromElements(                         Tuple2.of(＂key＂, 1),                         Tuple2.of(＂key＂, 2)                 );         FlinkJedisPoolConfig conf = new FlinkJedisPoolConfig.Builder().setHost(＂localhost＂).build();         stream.addSink(new RedisSink>(conf,new MyRedisMapper()));         String name = WriteToRedis.class.getName();         try {             env.execute(name);         }catch (Exception e)         {             e.printStackTrace();         }     } }