flink1.14消费Kafka存入redis

首先，需要使用Flink的Kafka Consumer连接到Kafka集群，并将数据读取到Flink的DataStream中。代码示例如下：

Properties props = new Properties();
props.setProperty("bootstrap.servers", "localhost:9092");
props.setProperty("group.id", "my-group");

FlinkKafkaConsumer<String> consumer = new FlinkKafkaConsumer<>("my-topic", new SimpleStringSchema(), props);
DataStream<String> kafkaStream = env.addSource(consumer);

接下来，可以使用Flink的redis-sink插件将数据写入Redis中。需要先添加redis-sink依赖，代码示例如下：

<dependency>
    <groupId>org.apache.flink</groupId>
    <artifactId>flink-connector-redis</artifactId>
    <version>1.14.0</version>
</dependency>

然后，可以使用RedisSink将数据写入Redis中。代码示例如下：

RedisCommandDescription redisCommandDescription = new RedisCommandDescription(RedisCommand.SET, null, 1);
RedisSink<String> redisSink = new RedisSink<>(jedisPoolConfig, new RedisMapper<String>() {
    @Override
    public RedisCommandDescription getCommandDescription() {
        return redisCommandDescription;
    }

    @Override
    public String getKeyFromData(String data) {
        // 从数据中获取key
        return "my-key";
    }

    @Override
    public String getValueFromData(String data) {
        // 从数据中获取value
        return data;
    }
});

kafkaStream.addSink(redisSink);

以上代码示例将Kafka中的数据写入Redis中，其中jedisPoolConfig为Redis连接池配置，可以根据实际情况进行配置。getKeyFromData和getValueFromData方法分别用于从数据中获取key和value，可以根据实际情况进行实现。