redis - 如何在flink map()中使用Jedis

Question

我的代码是这样的:

  StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

  env.enableCheckpointing(500);

  DataStream<String> stream = env.addSource(getConsumer(TOPIC_1));

  Jedis jedis = new Jedis("master1");
  stream.map(new RichMapFunction<String, String>() {
      @Override
      public String map(String value) throws Exception {
          String result = jedis.hget("rtc", value);
          return result;
      }
  });

我想在 map() 中从 Redis 获取一些数据，但是它无法运行，因为 Jedis.class 是不可序列化的。

如何在map()中使用ZkClient、Jedis等不可序列化的类？

Answer 1

所有 rich 函数(如 RichMapFunction)都有一个 open(Configuration) 和 close 调用，您可以覆盖它们.一旦将函数部署到执行函数的 TaskManager 中，就会调用这些生命周期方法。

class MyMapFunction extends RichMapFunction<String, String> {

    private transient Jedis jedis;

    @Override
    public void open(Configuration parameters) {
        // open connection to Redis, for example
        jedis = new Jedis("master1");
    }

    @Override
    public void close() {
        // close connection to Redis
        jedis.close();
    }
}