如何高效地将flink管道中的数据写入redis

myss37ts  于 2021-06-21  发布在  Flink
关注(0)|答案(2)|浏览(1018)

我正在ApacheFlinkSQLAPI中构建一个管道。管道执行简单的投影查询。但是,我需要在查询之前编写一次元组(确切地说是每个元组中的一些元素),在查询之后再编写一次。事实证明,我用来编写redis的代码严重降低了性能。i、 Flink以很小的数据速率产生反压力。我的代码有什么问题以及如何改进。有什么建议吗。
当我停止给redis写信时,前后的表现都非常出色。这是我的管道代码:

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<String> purchasesConsumer=new FlinkKafkaConsumer011<String>("purchases",
                new SimpleStringSchema(),
                props);

        DataStream<String> purchasesStream = env
                .addSource(purchasesConsumer)
                .setParallelism(Math.min(5 * 32, k_partitions));
        DataStream<Tuple4<Integer, Integer, Integer, Long>> purchaseWithTimestampsAndWatermarks =
                purchasesStream
                        .flatMap(new PurchasesParser())
                        .assignTimestampsAndWatermarks(new BoundedOutOfOrdernessTimestampExtractor<Tuple4<Integer, Integer, Integer, Long>>(Time.seconds(10)) {

                            @Override
                            public long extractTimestamp(Tuple4<Integer, Integer, Integer, Long> 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<Tuple2<Boolean, Row>> 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<Tuple4<Integer, Integer, Integer, Long>, String> {
    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<Integer, Integer, Integer, Long> input, Collector<String> 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);

    }
}

附加部分:我尝试了第二个实现,即使用jedis批量编写toredis的process函数。然而,我得到以下错误。org.apache.flink.runtime.client.jobexecutionexception:redis.clients.jedis.exceptions.jedisconnectionexception:java.net.socketexception:套接字未连接。我试图使成批消息的数量更小,但过了一段时间后仍然出现错误。
以下是流程功能的实现:
/**使用进程函数写入redis/

public static class WriteToRedisAfterQueryProcessFn extends ProcessFunction<Tuple2<Boolean, Row>, String> {
    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<Boolean, Row> input, Context context, Collector<String> 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();
        }
    }
}
u4dcyp6a

u4dcyp6a1#

通常在向外部服务写入时,这会成为flink工作流的瓶颈。提高性能的最简单方法是通过异步函数对工作流的这一部分进行多线程处理。有关详细信息,请参阅此文档。
--肯

j8yoct9x

j8yoct9x2#

毫无疑问,您遇到的糟糕性能是由于每次写入都向redis发出同步请求@kkrugler已经提到了异步i/o,这是解决这种情况的常见方法。这需要切换到支持异步操作的redis客户机之一。
处理外部服务时通常使用的另一种解决方案是将多组写入批处理在一起。对于绝地武士,你可以使用流水线。例如,您可以替换 WriteToRedis richflatmapfunction具有processfunction,该processfunction以一定大小的批量对redis执行流水线写入,并且根据需要依赖超时来刷新其缓冲区。你可以用flink的liststate作为缓冲区。

相关问题