package org.example.flink.operator;

import java.nio.charset.Charset;
import java.util.ArrayList;
import java.util.List;

import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;
import org.example.flink.data.Trace;

import com.google.common.hash.BloomFilter;
import com.google.common.hash.Funnels;


public class BloomFilterProcessFunction extends KeyedProcessFunction<String, Trace, Trace> {

	private static final long serialVersionUID = 1L;
	
	// bloom預(yù)計(jì)插入的數(shù)據(jù)量
	private static final long EXPECTED_INSERTIONS = 5000000L;
	// bloom的假陽性率
	private static final double FPP = 0.001;
	// bloom過濾器TTL
	private static final long TTL = 60 * 1000;
	// bloom過濾器隊(duì)列size
	private static final int FILTER_QUEUE_SIZE = 10;
	// bloom過濾器隊(duì)列
	private List<BloomFilter<String>> bloomFilterList;
	// 是否已經(jīng)注冊定時(shí)器
	private boolean registeredTimerTask = false;
	
	@Override
	public void open(Configuration parameters) throws Exception {
		bloomFilterList = new ArrayList<>(FILTER_QUEUE_SIZE);
		BloomFilter<String> bloomFilter = BloomFilter.create(Funnels.stringFunnel(Charset.forName("utf-8")),
				EXPECTED_INSERTIONS, FPP);
		bloomFilterList.add(bloomFilter);
	}
	
	@Override
	public void processElement(Trace trace, KeyedProcessFunction<String, Trace, Trace>.Context context,
			Collector<Trace> out) throws Exception {
		BloomFilter<String> firstBloomFilter = bloomFilterList.get(0);
		String key = trace.getGid();
		// 只要有一個(gè)bloom未hit該元素，就意味著該元素從未出現(xiàn)過，在隊(duì)列中的所有過濾器留下該元素的標(biāo)記
		if (!firstBloomFilter.mightContain(key)) {
			for (BloomFilter<String> bloomFilter : bloomFilterList) {
				bloomFilter.put(key);
			}
			// 該元素從未出現(xiàn)過，為非重復(fù)數(shù)據(jù)
			out.collect(trace);
		}
		if (!registeredTimerTask) {
			long current = context.timerService().currentProcessingTime();
			// 注冊處理時(shí)間定時(shí)器
			context.timerService().registerProcessingTimeTimer(current + TTL);
			registeredTimerTask = true;
		}
	}
	
	@Override
	public void onTimer(long timestamp, OnTimerContext context, Collector<Trace> out) throws Exception {
		// append新的bloomFilter到bloom過濾器隊(duì)列
		bloomFilterList
				.add(BloomFilter.create(Funnels.stringFunnel(Charset.forName("utf-8")), EXPECTED_INSERTIONS, FPP));
		// 清理第一個(gè)bloomFilter
		if (bloomFilterList.size() > FILTER_QUEUE_SIZE) {
			bloomFilterList.remove(0);
		}
		// 創(chuàng)建一個(gè)新的timer task
		context.timerService().registerProcessingTimeTimer(timestamp + TTL);
	}

	@Override
	public void close() throws Exception {
		bloomFilterList = null;
	}
}

package org.example.flink;

import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.connector.jdbc.JdbcConnectionOptions;
import org.apache.flink.connector.jdbc.JdbcExecutionOptions;
import org.apache.flink.connector.jdbc.JdbcSink;
import org.apache.flink.connector.kafka.source.KafkaSource;
import org.apache.flink.connector.kafka.source.enumerator.initializer.OffsetsInitializer;
import org.apache.flink.contrib.streaming.state.EmbeddedRocksDBStateBackend;
import org.apache.flink.streaming.api.datastream.DataStreamSink;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.example.flink.data.Trace;
import org.example.flink.operator.BloomFilterProcessFunction;

import com.google.gson.Gson;


public class StreamDeduplication {

	public static void main(String[] args) throws Exception {
		// 1. prepare
		Configuration configuration = new Configuration();
		configuration.setString("rest.port", "9091");
		StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(configuration);
		env.enableCheckpointing(2 * 60 * 1000);
		env.setStateBackend(new EmbeddedRocksDBStateBackend());  // 使用rocksDB作為狀態(tài)后端
		
		// 2. Kafka Source
		KafkaSource<String> source = KafkaSource.<String>builder()
			.setBootstrapServers("127.0.0.1:9092")
			.setTopics("trace")
			.setGroupId("group-01")
		    .setStartingOffsets(OffsetsInitializer.latest())
		    .setProperty("commit.offsets.on.checkpoint", "true")
		    .setValueOnlyDeserializer(new SimpleStringSchema())
		    .build();

		DataStreamSource<String> sourceStream = env.fromSource(source, WatermarkStrategy.noWatermarks(),
				"Kafka Source");
		sourceStream.setParallelism(1);	// 設(shè)置source算子的并行度為1
		
		// 3. 轉(zhuǎn)換為Trace對象
		SingleOutputStreamOperator<Trace> mapStream = sourceStream.map(new MapFunction<String, Trace>() {

			private static final long serialVersionUID = 1L;

			@Override
			public Trace map(String value) throws Exception {
				Gson gson = new Gson();
				Trace trace = gson.fromJson(value, Trace.class);
				return trace;
			}
		});
		mapStream.name("Map to Trace");
		mapStream.setParallelism(1);	// 設(shè)置map算子的并行度為1
		
		// 4. Bloom過濾器去重, 在去重之前要keyBy處理，保障同一gid的數(shù)據(jù)全都交由同一個(gè)線程處理
		SingleOutputStreamOperator<Trace> deduplicatedStream = mapStream.keyBy(
				new KeySelector<Trace, String>() {
					
					private static final long serialVersionUID = 1L;

					@Override
					public String getKey(Trace trace) throws Exception {
						return trace.getGid();
					}
				})
			.process(new BloomFilterProcessFunction());
		deduplicatedStream.name("Bloom filter process for distinct gid");
		deduplicatedStream.setParallelism(2);	// 設(shè)置去重算子的并行度為2
		
		// 5. 將去重結(jié)果寫入DataBase
		DataStreamSink<Trace> sinkStream = deduplicatedStream.addSink(
				JdbcSink.sink("insert into flink.deduplication(gid, timestamp) values (?, ?);",
						(statement, trace) -> {
							statement.setString(1, trace.getGid());
                            statement.setLong(2, trace.getTimestamp());
                        },
						JdbcExecutionOptions.builder()
                        		.withBatchSize(1000)
                        		.withBatchIntervalMs(200)
                        		.withMaxRetries(5)
                        		.build(), 
                        new JdbcConnectionOptions.JdbcConnectionOptionsBuilder()
                            	.withUrl("jdbc:mysql://127.0.0.1:3306/flink")
                            	.withUsername("username")
                            	.withPassword("password")
                            	.build())
				);
		sinkStream.name("Sink DB");
		sinkStream.setParallelism(1);
		
		// 執(zhí)行
		env.execute("Stream Real-Time Deduplication");
	}
}

public static void main(String[] args) throws InterruptedException {
    Properties props = new Properties();
    String topic = "trace";
    props.put("bootstrap.servers", "127.0.0.1:9092");
    props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
    props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
    Producer<String, String> producer = new KafkaProducer<String, String>(props);

    InputStream inputStream = KafkaDataProducer.class.getClassLoader().getResourceAsStream(TEST_DATA);
    Scanner scanner = new Scanner(inputStream, StandardCharsets.UTF_8.name());
    String content = scanner.useDelimiter("\\A").next();
    scanner.close();
    JSONObject jsonContent = JSONObject.parseObject(content);

    int nonDuplicateNum = 100000;
    int repeatNum = 100;
    Random r = new Random();
    for (int i = 0; i < nonDuplicateNum; i++) {
        String id = jsonContent.getString(GID);
        String newId = increase(id, String.valueOf(i));
        jsonContent.put(GID, newId);
        // 制造重復(fù)數(shù)據(jù)
        for (int j = 0; j < repeatNum; j++) {
            // 對時(shí)間進(jìn)行隨機(jī)擾動，模擬數(shù)據(jù)亂序到達(dá)
            long current = System.currentTimeMillis() - r.nextInt(60) * 1000;
            jsonContent.put(TIMESTAMP, current);
            producer.send(new ProducerRecord<String, String>(topic, jsonContent.toString()));
        }
        // wait some time
        Thread.sleep(5);
    }
    Thread.sleep(2000);
    System.out.println("\n");
    System.out.println("finished");
    producer.close();
}