RedisRichSinkFunction

自定義RedisRichSinkFunction，支持Redis數據庫連接池。

定義

public class RedisRichSinkFunction<IN> extends RichSinkFunction<IN> {
   private JedisPool pool = null;

   @Override
   public void open(Configuration parameters) throws Exception {
      ParameterTool parameterTool = (ParameterTool) getRuntimeContext().getExecutionConfig().getGlobalJobParameters();
      String hostname = parameterTool.get("redis_hostname");
      int port = Integer.parseInt(parameterTool.get("redis_port"));
      pool = RedisPoolFactory.getPool(hostname, port);
   }

   public JedisPool getPool() {
      return pool;
   }

}

使用

.addSink(new RedisRichSinkFunction[util.HashMap[String, (Integer, util.Set[String])]] {
 override def invoke(value: util.HashMap[String, (Integer, util.Set[String])], context: SinkFunction.Context[_]): Unit = {
  RedisHandler.setPvAndUv(getPool(), value)
 }
})