亿级数据如何分钟级别写入缓存？

result.foreachPartition(it -> {
            Jedis jedis = RedisInstance.getInstance(properties.getProperty("redis.ip"), Integer.parseInt(properties.getProperty("redis.port")), properties.getProperty("redis.pwd"));
            System.out.println(it.hashCode());
            Pipeline pipeline = jedis.pipelined();
            AtomicLong atomicLong = new AtomicLong();
            long start = System.currentTimeMillis();
            it.forEachRemaining(v -> {
                        //System.out.println(v.getString(0)+":"+v.getString(1));
                        atomicLong.incrementAndGet();
                        qpsControll(start, requiredQps, atomicLong, it.hashCode());
                        pipeline.sadd(v.getString(0), v.getString(1));
                        if (atomicLong.get() % 3 == 0) {
                            //每1000条提交一次
                            pipeline.sync();
                        }
                    }
            );
            pipeline.close();
            jedis.close();
        });

private static void qpsControll(long start, int requiredQps, AtomicLong count, int x) {
        //System.out.println("current count:"+x+":"+ count.get());
        long actualQps = 1000 * count.get() / (System.currentTimeMillis() - start);
        System.out.println(x + ":" + actualQps);
        if (actualQps > (long) requiredQps) {
            System.out.println("=====stop =====");
            try {
                Thread.sleep(1000);
            } catch (InterruptedException e) {
                System.out.println(e);
            }
        }
    }