title: How to write data from flink pipeline to redis efficiently
body: <p>I am building a pipeline in Apache flink sql api.
  The pipeline does simple projection query. However, I need to write the tuples (precisely some elements in the each tuple) once before the query and another time after the query.
  It turned out that my code that I am using to write to redis severely degrades performance. I.e the flink makes back pressure in a very small rate of data.
  What's wrong with my code and how can I improve. Any recommendations please. </p>
  
  <p>When I stopped writing to redis before and after  the performance was excellent. 
  Here is my pipeline code:</p>
  
  <pre><code>public class QueryExample {
      public static Long throughputCounterAfter=new Long("0");
      public static void main(String[] args) {
          int k_partitions = 10;
          reamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
          env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);
          env.setParallelism(5 * 32);
          Properties props = new Properties();
          props.setProperty("zookeeper.connect", "zookeeper-node-01:2181");
          props.setProperty("bootstrap.servers", "kafka-node-01:9092,kafka-node-02:9092,kafka-node-03:9092");
          // not to be shared with another job consuming the same topic
          props.setProperty("group.id", "flink-group");
          props.setProperty("enable.auto.commit","false");
          FlinkKafkaConsumer011&lt;String&gt; purchasesConsumer=new FlinkKafkaConsumer011&lt;String&gt;("purchases",
                  new SimpleStringSchema(),
                  props);
  
          DataStream&lt;String&gt; purchasesStream = env
                  .addSource(purchasesConsumer)
                  .setParallelism(Math.min(5 * 32, k_partitions));
          DataStream&lt;Tuple4&lt;Integer, Integer, Integer, Long&gt;&gt; purchaseWithTimestampsAndWatermarks =
                  purchasesStream
                          .flatMap(new PurchasesParser())
                          .assignTimestampsAndWatermarks(new BoundedOutOfOrdernessTimestampExtractor&lt;Tuple4&lt;Integer, Integer, Integer, Long&gt;&gt;(Time.seconds(10)) {
  
                              @Override
                              public long extractTimestamp(Tuple4&lt;Integer, Integer, Integer, Long&gt; element) {
                                  return element.getField(3);
                              }
                          });
  
          Table purchasesTable = tEnv.fromDataStream(purchaseWithTimestampsAndWatermarks, "userID, gemPackID,price, rowtime.rowtime");
          tEnv.registerTable("purchasesTable", purchasesTable);
  
          purchaseWithTimestampsAndWatermarks.flatMap(new WriteToRedis());
          Table result = tEnv.sqlQuery("SELECT  userID, gemPackID, rowtime from purchasesTable");
          DataStream&lt;Tuple2&lt;Boolean, Row&gt;&gt; queryResultAsDataStream = tEnv.toRetractStream(result, Row.class);
          queryResultAsDataStream.flatMap(new WriteToRedis());
          try {
              env.execute("flink SQL");
  
          } catch (Exception e) {
              e.printStackTrace();
          }
      }
  
  
  
  
  /**
   * write to redis
   */
  public static class WriteToRedis extends RichFlatMapFunction&lt;Tuple4&lt;Integer, Integer, Integer, Long&gt;, String&gt; {
      RedisReadAndWrite redisReadAndWrite;
  
      @Override
      public void open(Configuration parameters) {
          LOG.info("Opening connection with Jedis to {}", "redis");
          this.redisReadAndWrite = new RedisReadAndWrite("redis",6379);
  
      }
  
      @Override
      public void flatMap(Tuple4&lt;Integer, Integer, Integer, Long&gt; input, Collector&lt;String&gt; out) throws Exception {
          this.redisReadAndWrite.write(input.f0+":"+input.f3+"","time_seen", TimeUnit.NANOSECONDS.toMillis(System.nanoTime())+"");
      }
  }
  }
  
  
  public class RedisReadAndWrite {
      private Jedis flush_jedis;
  
      public RedisReadAndWrite(String redisServerName , int port) {
          flush_jedis=new Jedis(redisServerName,port);
      }
  
  
      public void write(String key,String field, String value) {
          flush_jedis.hset(key,field,value);
  
      }
  }
  </code></pre>
  
  <p>Additional part:
  I tried the second implementation the process function that batch the writing toredis using Jedis. However I am getting the following error. org.apache.flink.runtime.client.JobExecutionException: redis.clients.jedis.exceptions.JedisConnectionException: java.net.SocketException: Socket is not connected. I tried to make even the number of batched messages smaller and I am still getting errors after a while.</p>
  
  <p>Here is the implementation of the process function:</p>
  
  <p>/**
   * write to redis  using process function
   */</p>
  
  <pre><code>public static class WriteToRedisAfterQueryProcessFn extends ProcessFunction&lt;Tuple2&lt;Boolean, Row&gt;, String&gt; {
      Long timetoFlush;
      @Override
      public void open(Configuration parameters) {
          flush_jedis=new Jedis("redis",6379,1800);
          p = flush_jedis.pipelined();
          this.timetoFlush=System.currentTimeMillis()-initialTime;
      }
  
      @Override
      public void processElement(Tuple2&lt;Boolean, Row&gt; input, Context context, Collector&lt;String&gt; collector) throws Exception {
          p.hset(input.f1.getField(0)+":"+new Instant(input.f1.getField(2)).getMillis()+"","time_updated",TimeUnit.NANOSECONDS.toMillis(System.nanoTime())+"");
          throughputAccomulationcount++;
          System.out.println(throughputAccomulationcount);
          if(throughputAccomulationcount==50000){
              throughputAccomulationcount=0L;
              p.sync();
          }
      }
  }
  </code></pre>