Category, TimeStamp
Electronics,1679832334
Furniture,1679832336
Fashion,1679832378
Food,16798323536

package Aysnc_kafka_redis;

import AsyncIO.RedisSink;
import akka.japi.tuple.Tuple3;
import deserializer.Ecommdeserialize;
import model.Ecomm;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.connector.kafka.source.KafkaSource;
import org.apache.flink.connector.kafka.source.enumerator.initializer.OffsetsInitializer;
import org.apache.flink.streaming.api.datastream.AsyncDataStream;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.timestamps.BoundedOutOfOrdernessTimestampExtractor;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;
import java.util.concurrent.TimeUnit;

public class FlinkAsyncRedis {

    public static void main(String[] args) throws Exception {


        final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        Ecommdeserialize jsonde = new Ecommdeserialize(); 

        KafkaSource<Ecomm> source = KafkaSource.<Ecomm>builder()
                .setTopics("{dummytopic}")
                .setBootstrapServers("{dummybootstrap}")
                .setGroupId("test_flink")
                .setStartingOffsets(OffsetsInitializer.earliest())
                .setValueOnlyDeserializer(jsonde)
                .build();


        DataStream<Ecomm> orderData = env.fromSource(source, WatermarkStrategy.noWatermarks(), "Kafka Source");


        orderData.assignTimestampsAndWatermarks(new BoundedOutOfOrdernessTimestampExtractor<Ecomm>(Time.seconds(10)) {
            @Override
            public long extractTimestamp(Ecomm element) {
                return element.getEventTimestamp(); // extract watermark column from stream
            }
        });

        SingleOutputStreamOperator<Tuple3<String, Long, Long>> aggregatedData = orderData.keyBy(Ecomm::getCategory)
                .window(SlidingEventTimeWindows.of(Time.hours(24),Time.seconds(30)))
                .apply((WindowFunction<Ecomm, Tuple3<String, Long, Long>, String, TimeWindow>) (key, window, input, out) -> {
                    long count = 0;
                    for (Ecomm event : input) {
                        count++; // increment the count for each event in the window
                    }
                    out.collect(new Tuple3<>(key, window.getEnd(), count)); // output the category, window end time, and count
                });


        // calling async I/0 operator to sink data to redis in UnOrdered way
        SingleOutputStreamOperator<String> sinkResults = AsyncDataStream.unorderedWait(aggregatedData,new RedisSink(
                "{redisClusterUrl}"),
                1000, // the timeout defines how long an asynchronous operation take before it is finally considered failed
                TimeUnit.MILLISECONDS,
                 100); //capacity This parameter defines how many asynchronous requests may be in progress at the same time.

        sinkResults.print(); // print out the redis set response stored in the future for every key

        env.execute("RedisAsyncSink"); // you will be able to see your job running on cluster by this name


    }

}

package AsyncIO;

import akka.japi.tuple.Tuple3;
import io.lettuce.core.RedisFuture;
import io.lettuce.core.cluster.RedisClusterClient;
import io.lettuce.core.cluster.api.StatefulRedisClusterConnection;
import io.lettuce.core.cluster.api.async.RedisAdvancedClusterAsyncCommands;
import lombok.AllArgsConstructor;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.async.ResultFuture;
import org.apache.flink.streaming.api.functions.async.RichAsyncFunction;
import scala.collection.immutable.List;

import java.util.ArrayList;
import java.util.Collections;

@AllArgsConstructor
public class RedisSink extends RichAsyncFunction<Tuple3<String, Long, Long>, String> {

    String redisUrl;

    public RedisSink(String redisUrl){
        this.redisUrl=redisUrl;
    }

    private transient RedisClusterClient client = null;
    private transient StatefulRedisClusterConnection<String, String> clusterConnection = null;
    private transient RedisAdvancedClusterAsyncCommands<String, String> asyncCall = null;


    // method executes any operator-specific initialization 
    @Override
    public void open(Configuration parameters) {
        if (client == null ) {
            client = RedisClusterClient.create(redisUrl);
        }
        if (clusterConnection == null) {
            clusterConnection = client.connect();
        }
        if (asyncCall == null) {
            asyncCall  = clusterConnection.async();
        }
    }

    // core logic to set key in redis using async connection and return result of the call via ResultFuture
    @Override
    public void asyncInvoke(Tuple3<String, Long, Long> stream, ResultFuture<String> resultFuture) {

        String productKey = stream.t1();
        System.out.println("RedisKey:" + productKey); //for logging
        String count = stream.t3().toString();
        System.out.println("Redisvalue:" +  count); //for logging
        RedisFuture<String> setResult = asyncCall.set(productKey,count);

        setResult.whenComplete((result, throwable) -> {if(throwable!=null){
            System.out.println("Callback from redis failed:" + throwable);
            resultFuture.complete(new ArrayList<>());
        }
        else{
            resultFuture.complete(new ArrayList(Collections.singleton(result)));
        }});
    }
    
     // method closes what was opened during initialization to free any resources 
    //  held by the operator (e.g. open network connections, io streams)
    @Override
    public void close() throws Exception {
        client.close();
    }

}