/********************************** 数据去重 **************************************/
//以前没有记录偏移量，就从头读，如果记录过偏移量，就接着读
properties.setProperty("auto.offset.reset", "earliest");
//不自动提交偏移量，让flink提交偏移量
properties.setProperty("enable.auto.commit", "false");
KeyedStream<KafkaEvent, Tuple> keyed = env.addSource(new FlinkKafkaConsumer011<String>(
parameterTool.getRequired("input-topic"),
new SimpleStringSchema(),
parameterTool.getProperties()
)).map(new MapFunction<String, KafkaEvent>() {
@Override
public KafkaEvent map(String line) throws Exception {
return KafkaEvent.fromString(line);
}
}).filter(new FilterFunction<KafkaEvent>() {
@Override
public boolean filter(KafkaEvent value) throws Exception {
return value.getTpp()>=360;
}
}).keyBy("si", "ci");
SingleOutputStreamOperator<String> mapstatedata = keyed.process(new ProcessFunction<KafkaEvent, String>() {
private transient ValueState<BloomFilter> bloomState;
//private transient ValueState<Integer> counterState;
@Override
public void open(Configuration parameters) throws Exception {
//定义一个布隆过滤器的状态描述器
ValueStateDescriptor<BloomFilter> bloomStateDescriptor = new ValueStateDescriptor<>("bloomState", BloomFilter.class);
//ListStateDescriptor<BloomFilter> bloomStateDescriptor = new ListStateDescriptor<>("bloomState", BloomFilter.class);
//ValueStateDescriptor<Integer> counterDescriptor = new ValueStateDescriptor<>("counterState", Integer.class);
bloomState = getRuntimeContext().getState(bloomStateDescriptor);
//bloomState = getRuntimeContext().getListState(bloomStateDescriptor);
//counterState = getRuntimeContext().getState(counterDescriptor);
}
@Override
public void processElement(KafkaEvent value, Context context, Collector<String> out) throws Exception {
String tno = value.getTno();
BloomFilter bloomFilter = bloomState.value();
StringBuilder stringBuilder = new StringBuilder();
if (bloomState.value() == null) {
bloomFilter = BloomFilter.create(Funnels.unencodedCharsFunnel(), 100000);
//counterState.update(0);
}
//返回false代表一定不存在
if (!bloomFilter.mightContain(tno)) {
bloomFilter.put(tno);
bloomState.update(bloomFilter);
//counterState.update(counterState.value() + 1);
//System.out.println("countstate ==> "+counterState.value());
String result = stringBuilder//.append(counterState.value()).append("^|")
.append(value.getDid()).append("^|")
.append(value.getUvs()).toString();
out.collect(result);
}
}
})/*.uid("mapStateUid")*/.setParallelism(1);
/**
* 全局row_num，operator 状态管理
*/
public static class OperatorStateMap extends RichFlatMapFunction<String, Tuple2<Long, String>> implements CheckpointedFunction {
//托管状态
private transient ListState<Long> listState;
//原始状态
//private List<Long> listElements = new CopyOnWriteArrayList<>();
Long listElement = 0L;
Long i = 0L;
@Override
public void flatMap(String value, Collector collector) throws Exception {
i = listElement;
if(i == 0) {
i = i + 1;
collector.collect(new Tuple2<Long, String>(i, i + "|^|" + value));
listElement = i;
}else{
listElement = listElement + 1;
i++;
collector.collect(new Tuple2<Long, String>(listElement, listElement + "|^|" + value));
}
//listElements.clear();
//listElements.add(i);
}
/**
* 进行checkpoint进行快照
*
* @param context
* @throws Exception
*/
@Override
public void snapshotState(FunctionSnapshotContext context) throws Exception {
listState.clear();
/*for (Long ele : listElements) {
System.out.println("===============>snapshot存储原始状态"+ele);
listState.add(ele);
}*/
listState.add(listElement);
}
/**
* state的初始状态，包括从故障恢复过来
*
* @param context
* @throws Exception
*/
@Override
public void initializeState(FunctionInitializationContext context) throws Exception {
ListStateDescriptor listStateDescriptor = new ListStateDescriptor("checkPointedList",
TypeInformation.of(new TypeHint<Long>() {
}));
listState = context.getOperatorStateStore().getListState(listStateDescriptor);
//如果是故障恢复
/*if (context.isRestored()) {
//从托管状态将数据到移动到原始状态
for (Long ele : listState.get()) {
listElements.add(ele);
}
//listState.clear();
}*/
if (context.isRestored()) {
Iterator<Long> iterator = listState.get().iterator();
while (iterator.hasNext()){
Long ele = iterator.next();
System.out.println("===============>"+ele);
//listElements.add(ele);
listElement = ele;
}
}
}
@Override
public void open(Configuration parameters) throws Exception {
//listElements = new CopyOnWriteArrayList<>();
}
}
}
