apache flink - Fink에서 실시간으로 가치를보고하는 방법은 무엇입니까?

더 자주 업데이트하려면 QueryableState 를 사용하십시오 , 사용 사례에 적합한 속도로 상태를 폴링합니다.

ContinuousEventTimeTrigger를 사용하면 전체 창보다 짧은 시간에 창이 실행되어 중간 상태를 볼 수 있습니다. 싱크의 다운 스트림 소비자가 각 출력이 전체 전류 상태가 아닌 부분 집계가 될 것으로 예상하는 경우 PurgingTrigger에서 선택적으로 래핑 할 수 있습니다.

CEP를 시도했습니다.

코드 :

AfterMatchSkipStrategy strategy = AfterMatchSkipStrategy.skipShortOnes();
    Pattern<RiskEvent, ?> loginPattern = Pattern.<RiskEvent>begin("start", strategy)
            .where(eventTypeCondition)
            .timesOrMore(1)
            .greedy()
            .within(Time.hours(1));

    KeyedStream<RiskEvent, String> keyedStream = dataStream.keyBy(new KeySelector<RiskEvent, String>() {
        @Override
        public String getKey(RiskEvent riskEvent) throws Exception {
            // key by user for aggregation
            return riskEvent.getEventType() + riskEvent.getDeviceFp();
        }
    });
    PatternStream<RiskEvent> eventPatternStream = CEP.pattern(keyedStream, loginPattern);
    eventPatternStream.select(new PatternSelectFunction<RiskEvent, RiskResult>() {
        @Override
        public RiskResult select(Map<String, List<RiskEvent>> map) throws Exception {
            List<RiskEvent> list = map.get("start");
            ArrayList<Long> times = new ArrayList<>();
            for (RiskEvent riskEvent : list) {
                times.add(riskEvent.getEventTime());
            }
            Long min = Collections.min(times);
            Long max = Collections.max(times);
            Set<String> accountList = list.stream().map(RiskEvent::getUserName).collect(Collectors.toSet());
            logger.info("时间范围：" + new Date(min) + " --- " + new Date(max) + " 事件:" + list.get(0).getEventType() + ", 设备指纹:" + list.get(0).getDeviceFp() + ", 关联账户:" + accountList.toString());
            return null;
        }
    });

아마도, 스킵 전략 skipShortOnes  맞춤 전략입니다.

CEP 라이브러리의 수정 사항을 보여주십시오.

<올>

Enum에서 전략 추가

공개 열거 형 SkipStrategy {     NO_SKIP,     SKIP_PAST_LAST_EVENT,     SKIP_TO_FIRST,     SKIP_TO_LAST,     SKIP_SHORT_ONES }

AfterMatchSkipStrategy.java 의 액세스 방법 추가

공개 정적 AfterMatchSkipStrategy skipShortOnes () {     새 AfterMatchSkipStrategy (SkipStrategy.SKIP_SHORT_ONES)를 반환합니다. }

discardComputationStatesAccordingToStrategy 에서 전략 액션 추가   NFA.java 의 방법 .

케이스 SKIP_SHORT_ONES :             int i = 0;             List >>tempResult = new ArrayList<>(일치 결과);             for (Map>resultMap : tempResult) {                 if (i ++ == 0) {                     계속하다;                 }                 matchedResult.remove (resultMap);             }             휴식;