Я строю конвейер в Apache flink sql api.Конвейер выполняет простой проекционный запрос.Тем не менее, мне нужно написать кортежи (точно некоторые элементы в каждом кортеже) один раз перед запросом и в другой раз после запроса.Оказалось, что мой код, который я использую для записи в Redis, сильно снижает производительность.Т.е. флинк оказывает обратное давление при очень малой скорости передачи данных.Что не так с моим кодом и как я могу улучшить.Любые рекомендации, пожалуйста.
Когда я перестал писать в 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);
}
}
Дополнительная часть: я попробовал вторую реализацию функции процесса, которая пакетирует запись toredis с использованием Jedis.Однако я получаю следующую ошибку.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();
}
}
}