flink写入redis

太过爱你忘了你带给我的痛 2023-06-09 10:04 23阅读 0赞

## 使用 Flink Redis Connector ##

支持三种Redis连接方式

1.  **Single Redis Server**
2.  **Redis Cluster**
3.  **Redis Sentinel**

源码：[https://github.com/apache/bahir-flink][https_github.com_apache_bahir-flink]

添加依赖：

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

如果找不到对应的依赖，可以使用如下依赖：

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

### 使用 Single Redis Server： ###

FlinkJedisPoolConfig conf = new FlinkJedisPoolConfig.Builder().setHost("127.0.0.1").build();
    
    DataStream<String> stream = ...;
    stream.addSink(new RedisSink<Tuple2<String, String>>(conf, new RedisExampleMapper());

其中 RedisExampleMapper：

public static class RedisExampleMapper implements RedisMapper<Tuple2<String, String>>{
    
        @Override
        public RedisCommandDescription getCommandDescription() {
            return new RedisCommandDescription(RedisCommand.HSET, "HASH_NAME");
        }
    
        @Override
        public String getKeyFromData(Tuple2<String, String> data) {
            return data.f0;
        }
    
        @Override
        public String getValueFromData(Tuple2<String, String> data) {
            return data.f1;
        }
    }

重写以下三个方法即可：

getCommandDescription() 返回对应Redis命令

getKeyFromData() 从数据中获取对应的key

getValueFromData()从数据中获取对应的value

### 使用 Redis Cluster： ###

FlinkJedisPoolConfig conf = new FlinkJedisPoolConfig.Builder()
        .setNodes(new HashSet<InetSocketAddress>(Arrays.asList(new InetSocketAddress(5601)))).build();
    
    DataStream<String> stream = ...;
    stream.addSink(new RedisSink<Tuple2<String, String>>(conf, new RedisExampleMapper());

###  使用 Redis Sentinel： ###

FlinkJedisSentinelConfig conf = new FlinkJedisSentinelConfig.Builder()
        .setMasterName("master").setSentinels(...).build();
    
    DataStream<String> stream = ...;
    stream.addSink(new RedisSink<Tuple2<String, String>>(conf, new RedisExampleMapper());

**以上，操作简单，不过限制比较多，很多redis的操作不支持。**

**如果需要的话，根据源码写一版RedisSink继承RichSinkFunction，详见：**

**[https://blog.csdn.net/www\_changer/article/details/102654195][https_blog.csdn.net_www_changer_article_details_102654195]**

[https_github.com_apache_bahir-flink]: https://github.com/apache/bahir-flink
[https_blog.csdn.net_www_changer_article_details_102654195]: https://blog.csdn.net/www_changer/article/details/102654195