[Flink]Flink章3 Flink应用开发 --- Redis Sink

Flink提供了专门操作redis的Redis Sink

依赖

 <dependency>
    <groupId>org.apache.bahir</groupId>
    <artifactId>flink-connector-redis_2.11</artifactId>
    <version>1.0</version>
</dependency>

类

Redis Sink 提供用于向Redis发送数据的接口的类。接收器可以使用三种不同的方法与不同类型的Redis环境进行通信：

类	场景	备注
FlinkJedisPoolConfig	单Redis服务器	适用于本地、测试场景
FlinkJedisClusterConfig	Redis集群
FlinkJedisSentinelConfig	Redis哨兵

 

使用

Redis Sink 核心类是 RedisMappe 是一个接口，使用时我们要编写自己的redis操作类实现这个接口中的三个方法

RedisMapper

public interface RedisMapper<T> extends Function, Serializable {

    /**
     * 设置使用的redis数据结构类型，和key的名词
     * 通过RedisCommand设置数据结构类型
     * Returns descriptor which defines data type.
     *
     * @return data type descriptor
     */
    RedisCommandDescription getCommandDescription();

    /**
     * 设置value中的键值对 key的值
     * Extracts key from data.
     *
     * @param data source data
     * @return key
     */
    String getKeyFromData(T data);

    /**
     * 设置value中的键值对 value的值
     * Extracts value from data.
     *
     * @param data source data
     * @return value
     */
    String getValueFromData(T data);
}

RedisCommand

使用RedisCommand设置数据结构类型时和redis结构对应关系。

Data Type	Redis Command [Sink]
HASH	HSET
LIST	RPUSH, LPUSH
SET	SADD
PUBSUB	PUBLISH
STRING	SET
HYPER_LOG_LOG	PFADD
SORTED_SET	ZADD
SORTED_SET	ZREM

Demo

public class RedisSinkTest { public static void main(String[] args) throws Exception{ StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(); env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime); env.enableCheckpointing(2000); env.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE); //连接kafka Properties properties = new Properties(); properties.setProperty("bootstrap.servers", "127.0.0.1:9092"); FlinkKafkaConsumer<String> consumer = new FlinkKafkaConsumer<>("test", new SimpleStringSchema(), properties); consumer.setStartFromEarliest(); DataStream<String> stream = env.addSource(consumer); DataStream<Tuple2<String, Integer>> counts = stream.flatMap(new LineSplitter()).keyBy(0).sum(1); //实例化FlinkJedisPoolConfig 配置redis FlinkJedisPoolConfig conf = new FlinkJedisPoolConfig.Builder().setHost("127.0.0.1").setPort("6379").build(); //实例化RedisSink，并通过flink的addSink的方式将flink计算的结果插入到redis counts.addSink(new RedisSink<>(conf,new RedisSinkExample())); env.execute("WordCount From Kafka To Redis"); } public static final class LineSplitter implements FlatMapFunction<String, Tuple2<String, Integer>> { @Override public void flatMap(String value, Collector<Tuple2<String, Integer>> out) { String[] tokens = value.toLowerCase().split("\\W+"); for (String token : tokens) { if (token.length() > 0) { out.collect(new Tuple2<String, Integer>(token, 1)); } } } } //指定Redis set public static final class RedisSinkExample implements RedisMapper<Tuple2<String,Integer>> { public RedisCommandDescription getCommandDescription() { return new RedisCommandDescription(RedisCommand.SET, null); } public String getKeyFromData(Tuple2<String, Integer> data) { return data.f0; } public String getValueFromData(Tuple2<String, Integer> data) { return data.f1.toString(); } } }