FlinkCEP – Flink의 복합 이벤트 처리

FlinkCEP – Flink의 복합 이벤트 처리 (CEP)

FlinkCEP는 Flink 위에 구현된 복합 이벤트 처리(Complex Event Processing, CEP) 라이브러리예요. 끝없이 흐르는 이벤트 스트림에서 이벤트 패턴을 감지해서, 데이터에서 중요한 부분을 놓치지 않고 잡아낼 수 있게 해줘요.

이 페이지에서는 Flink CEP에서 제공하는 API 호출을 다룹니다. 먼저 스트림에서 감지하고 싶은 패턴을 정의하는 Pattern API를 소개하고, 그다음 매칭된 이벤트 시퀀스를 어떻게 감지하고 처리하는지 보여줘요. 이어서 CEP 라이브러리가 이벤트 타임에서 지연(lateness)을 처리하는 방식과, 이전 Flink 버전에서 잡을 마이그레이션하는 방법을 다룹니다.

출처: Event Processing (CEP)

본문

FlinkCEP는 Flink가 제공하는 복합 이벤트 처리(CEP) 라이브러리예요. 이 라이브러리로 이벤트 패턴을 스트림에서 감지하여 데이터에서 중요한 부분을 찾아낼 수 있어요.

바로 시작하고 싶다면 Flink 프로그램을 설정하고 프로젝트의 pom.xml에 FlinkCEP 의존성을 추가하세요.

<dependency>
  <groupId>org.apache.flink</groupId>
  <artifactId>flink-cep</artifactId>
  <version>${flink.version}</version>
</dependency>

FlinkCEP는 바이너리 배포판의 일부가 아니에요. 클러스터 실행을 위해 연결하는 방법은 여기를 참고하세요.

이제 Pattern API를 사용해 첫 CEP 프로그램을 작성해볼 수 있어요.

경고: 패턴 매칭을 적용할 DataStream의 이벤트는 올바른 equals()와 hashCode() 메서드를 구현해야 해요. FlinkCEP가 이벤트를 비교·매칭할 때 이 메서드를 사용하기 때문이에요.

DataStream<Event> input = ...;

Pattern<Event, ?> pattern = Pattern.<Event>begin("start")
    .where(SimpleCondition.of(event -> event.getId() == 42))
    .next("middle")
    .subtype(SubEvent.class)
    .where(SimpleCondition.of(subEvent -> subEvent.getVolume() >= 10.0))
    .followedBy("end")
    .where(SimpleCondition.of(event -> event.getName().equals("end")));

PatternStream<Event> patternStream = CEP.pattern(input, pattern);

DataStream<Alert> result = patternStream.process(
    new PatternProcessFunction<Event, Alert>() {
        @Override
        public void processMatch(
                Map<String, List<Event>> pattern,
                Context ctx,
                Collector<Alert> out) throws Exception {
            out.collect(createAlertFrom(pattern));
        }
    });

The Pattern API

Pattern API는 입력 스트림에서 추출하고 싶은 복잡한 패턴 시퀀스를 정의할 수 있게 해줘요.

각 복합 패턴 시퀀스는 여러 개의 단순 패턴, 즉 같은 속성을 가진 개별 이벤트를 찾는 패턴으로 구성돼요. 이제부터 이 단순 패턴을 패턴(pattern), 스트림에서 찾고 있는 최종 복합 패턴 시퀀스를 **패턴 시퀀스(pattern sequence)**라고 부를게요. 패턴 시퀀스는 그런 패턴들의 그래프로 볼 수 있는데, 한 패턴에서 다음 패턴으로의 전이는 사용자가 지정한 조건 예를 들어 event.getName().equals("end")에 따라 발생해요. **매치(match)**는 유효한 패턴 전이를 따라 복합 패턴 그래프의 모든 패턴을 방문하는 입력 이벤트의 시퀀스예요.

정보: 각 패턴은 고유한 이름을 가져야 해요. 이 이름은 나중에 매칭된 이벤트를 식별하는 데 사용돼요.

위험: 패턴 이름은 ":" 문자를 포함할 수 없어요.

이 섹션의 나머지에서는 먼저 개별 패턴(Individual Patterns)을 정의하는 방법을 설명하고, 그다음 개별 패턴을 복합 패턴(Complex Patterns)으로 결합하는 방법을 다룰게요.

Individual Patterns

Pattern은 싱글턴(singleton) 또는 루핑(looping) 패턴이 될 수 있어요. 싱글턴 패턴은 단일 이벤트를 받아들이고, 루핑 패턴은 두 개 이상을 받아들일 수 있어요. 패턴 매칭 기호로 "a b+ c? d"(즉 "a", 그리고 하나 이상의 "b", 선택적으로 "c", 그리고 "d")라는 패턴에서 a, c?, d는 싱글턴 패턴이고 b+는 루핑 패턴이에요. 기본적으로 패턴은 싱글턴이며, Quantifiers를 사용해 루핑 패턴으로 바꿀 수 있어요. 각 패턴은 이벤트를 받아들이는 기준이 되는 Conditions을 하나 이상 가질 수 있어요.

Quantifiers

FlinkCEP에서 루핑 패턴은 다음 메서드로 지정해요: 주어진 이벤트가 한 번 이상 발생할 것으로 기대하는 패턴에는 pattern.oneOrMore()(앞서 언급한 b+ 같은 경우), 특정 유형의 이벤트가 정확히 몇 번 발생할 것으로 기대하는 패턴에는 pattern.times(#ofTimes)(예: a 4개), 주어진 유형 이벤트의 최소·최대 발생 횟수를 기대하는 패턴에는 pattern.times(#fromTimes, #toTimes)(예: a 2~4개)를 사용해요.

pattern.greedy() 메서드로 루핑 패턴을 탐욕적으로(greedy) 만들 수 있지만, 그룹 패턴은 아직 탐욕적으로 만들 수 없어요. 그리고 루핑 여부와 무관하게 모든 패턴을 pattern.optional() 메서드로 선택적으로 만들 수 있어요.

start라는 이름의 패턴에 대해 유효한 quantifier는 다음과 같아요:

// 4회 발생 기대
start.times(4);

// 0회 또는 4회 발생 기대
start.times(4).optional();

// 2, 3 또는 4회 발생 기대
start.times(2, 4);

// 2, 3 또는 4회 발생, 가능한 많이 반복
start.times(2, 4).greedy();

// 0, 2, 3 또는 4회 발생 기대
start.times(2, 4).optional();

// 0, 2, 3 또는 4회 발생, 가능한 많이 반복
start.times(2, 4).optional().greedy();

// 1회 이상 발생 기대
start.oneOrMore();

// 1회 이상 발생, 가능한 많이 반복
start.oneOrMore().greedy();

// 0회 이상 발생 기대
start.oneOrMore().optional();

// 0회 이상 발생, 가능한 많이 반복
start.oneOrMore().optional().greedy();

// 2회 이상 발생 기대
start.timesOrMore(2);

// 2회 이상 발생, 가능한 많이 반복
start.timesOrMore(2).greedy();

// 0, 2회 이상 발생 기대
start.timesOrMore(2).optional()

// 0, 2회 이상 발생, 가능한 많이 반복
start.timesOrMore(2).optional().greedy();

Conditions

각 패턴에 대해 들어오는 이벤트가 패턴에 "수용"되기 위해 충족해야 하는 조건을 지정할 수 있어요. 예를 들어 값이 5보다 커야 한다거나, 이전에 수용된 이벤트들의 평균값보다 커야 한다는 식이에요. 이벤트 속성에 대한 조건은 pattern.where(), pattern.or() 또는 pattern.until() 메서드로 지정할 수 있고, 이들은 IterativeCondition이거나 SimpleCondition일 수 있어요.

Iterative Conditions: 가장 일반적인 유형의 조건이에요. 이전에 수용된 이벤트의 속성이나 그 일부에 대한 통계를 기반으로 이후 이벤트를 수용하는 조건을 지정하는 방법이에요.

아래는 "middle"이라는 이름의 패턴에 대해, 이름이 "foo"로 시작하고 해당 패턴에 대해 이전에 수용된 이벤트들의 가격 합에 현재 이벤트의 가격을 더한 값이 5.0을 넘지 않으면 다음 이벤트를 수용하는 반복 조건 코드예요. 반복 조건은 특히 oneOrMore() 같은 루핑 패턴과 함께 쓸 때 강력해요.

middle.oneOrMore()
    .subtype(SubEvent.class)
    .where(new IterativeCondition<SubEvent>() {
        @Override
        public boolean filter(SubEvent value, Context<SubEvent> ctx) throws Exception {
            if (!value.getName().startsWith("foo")) {
                return false;
            }

            double sum = value.getPrice();
            for (Event event : ctx.getEventsForPattern("middle")) {
                sum += event.getPrice();
            }
            return Double.compare(sum, 5.0) < 0;
        }
    });

정보: ctx.getEventsForPattern(...) 호출은 주어진 잠재 매치에 대해 이전에 수용된 모든 이벤트를 찾아줘요. 이 연산의 비용은 달라질 수 있으므로 조건을 구현할 때는 사용을 최소화하려고 노력하세요.

설명된 컨텍스트는 이벤트 타임 특성에도 접근을 제공해요. 자세한 내용은 Time context를 참고하세요.

Simple Conditions: 이 유형의 조건은 앞서 언급한 IterativeCondition 클래스를 확장하며, 오직 이벤트 자체의 속성만을 기반으로 이벤트를 수용할지 결정해요.

start.where(SimpleCondition.of(value -> value.getName().startsWith("foo")));

마지막으로 pattern.subtype(subClass) 메서드를 통해 수용되는 이벤트의 유형을 초기 이벤트 유형(여기서는 Event)의 하위 유형으로 제한할 수도 있어요.

start.subtype(SubEvent.class)
        .where(SimpleCondition.of(value -> ... /*some condition*/));

Combining Conditions: 위에서 보여줬듯이 subtype 조건을 추가 조건과 결합할 수 있어요. 이는 모든 조건에 적용돼요. where()를 순차적으로 호출해 조건을 자유롭게 결합할 수 있어요. 최종 결과는 개별 조건 결과들의 논리적 AND가 돼요. OR를 사용해 조건을 결합하려면 아래와 같이 or() 메서드를 사용할 수 있어요.

pattern.where(SimpleCondition.of(value -> ... /*some condition*/))
        .or(SimpleCondition.of(value -> ... /*some condition*/));

Stop condition: 루핑 패턴(oneOrMore()와 oneOrMore().optional())에서는 중지 조건도 지정할 수 있어요. 예를 들어 값 합이 50보다 작을 때까지 값이 5보다 큰 이벤트를 수용하는 식이에요.

더 잘 이해하기 위해 다음 예시를 볼게요. 주어진

• 패턴 "(a+ until b)" (하나 이상의 "a"이고 "b"까지)

• 들어오는 이벤트 시퀀스 "a1" "c" "a2" "b" "a3"

• 라이브러리는 결과: {a1 a2} {a1} {a2} {a3}을 출력해요.

보시다시피 중지 조건 때문에 {a1 a2 a3}이나 {a2 a3}은 반환되지 않아요.

where(condition)

현재 패턴에 대한 조건을 정의해요. 패턴을 매치하려면 이벤트가 조건을 충족해야 해요. 연속된 where() 절은 그 조건들이 AND로 결합되게 해요.

pattern.where(new IterativeCondition<Event>() {
    @Override
    public boolean filter(Event value, Context ctx) throws Exception {
        return ...; // some condition
    }
});

or(condition)

기존 조건과 OR로 결합되는 새 조건을 추가해요. 이벤트가 패턴을 매치하려면 조건 중 적어도 하나를 통과해야 해요.

pattern.where(new IterativeCondition<Event>() {
    @Override
    public boolean filter(Event value, Context ctx) throws Exception {
        return ...; // some condition
    }
}).or(new IterativeCondition<Event>() {
    @Override
    public boolean filter(Event value, Context ctx) throws Exception {
        return ...; // alternative condition
    }
});

until(condition)

루핑 패턴에 대한 중지 조건을 지정해요. 주어진 조건과 일치하는 이벤트가 발생하면 더 이상 이벤트가 패턴에 수용되지 않는다는 뜻이에요. oneOrMore()와 함께만 사용 가능해요. 참고: 이벤트 기반 조건에 해당 패턴의 상태를 정리하는 것을 허용해요.

pattern.oneOrMore().until(new IterativeCondition<Event>() {
    @Override
    public boolean filter(Event value, Context ctx) throws Exception {
        return ...; // alternative condition
    }
});

subtype(subClass)

현재 패턴에 대한 하위 유형 조건을 정의해요. 이벤트는 이것이 이 하위 유형일 때만 패턴을 매치할 수 있어요.

pattern.subtype(SubEvent.class);

oneOrMore()

이 패턴이 매칭 이벤트의 최소 한 번의 발생을 기대한다고 지정해요. 기본적으로 (연속 이벤트 사이의) 완화된 내부 연속성(relaxed internal contiguity)이 사용돼요. 내부 연속성에 대한 자세한 내용은 consecutive를 참고하세요. 상태 정리를 활성화하려면 until()이나 within()을 사용하는 것이 좋아요.

pattern.oneOrMore();

timesOrMore(#times)

이 패턴이 매칭 이벤트의 최소 #times번 발생을 기대한다고 지정해요. 기본적으로 완화된 내부 연속성이 사용돼요. 자세한 내용은 consecutive 참고.

pattern.timesOrMore(2);

times(#ofTimes)

이 패턴이 매칭 이벤트의 정확한 발생 횟수를 기대한다고 지정해요. 기본값은 완화된 내부 연속성. 자세한 내용은 consecutive 참고.

pattern.times(2);

times(#fromTimes, #toTimes)

이 패턴이 매칭 이벤트의 #fromTimes에서 #toTimes 사이의 발생을 기대한다고 지정해요. 기본값은 완화된 내부 연속성. 자세한 내용은 consecutive 참고.

pattern.times(2, 4);

optional()

이 패턴이 선택적(optional), 즉 전혀 발생하지 않을 수도 있다고 지정해요. 앞서 언급한 모든 quantifier에 적용 가능해요.

pattern.oneOrMore().optional();

greedy()

이 패턴이 탐욕적(greedy), 즉 가능한 한 많이 반복한다고 지정해요. 이는 quantifier에만 적용되며 현재 그룹 패턴은 지원하지 않아요.

pattern.oneOrMore().greedy();

Combining Patterns

개별 패턴이 어떤 모습인지 봤으니 이제 그것들을 전체 패턴 시퀀스로 결합하는 방법을 볼게요.

패턴 시퀀스는 아래와 같이 초기 패턴으로 시작해야 해요:

Pattern<Event, ?> start = Pattern.<Event>begin("start");

다음으로 패턴 시퀀스에 패턴을 추가하려면 이벤트 사이의 원하는 *연속성 조건(contiguity conditions)*을 지정하면 돼요. FlinkCEP는 이벤트 사이에 다음 형태의 연속성을 지원해요:

1. Strict Contiguity(엄격 연속성): 모든 매칭 이벤트가 중간에 비매칭 이벤트 없이 반드시 연속적으로 나타나야 해요.

2. Relaxed Contiguity(완화 연속성): 매칭 이벤트 사이에 나타나는 비매칭 이벤트를 무시해요.

3. Non-Deterministic Relaxed Contiguity(비결정적 완화 연속성): 연속성을 더 완화해, 일부 매칭 이벤트를 무시하는 추가 매치를 허용해요.

이를 연속된 패턴 사이에 적용하려면 다음을 사용할 수 있어요:

1. next() — strict,
2. followedBy() — relaxed,
3. followedByAny() — non-deterministic relaxed 연속성.

또는

1. notNext() — 어떤 이벤트 유형이 다른 것을 바로 뒤따르지 않게 하고 싶을 때
2. notFollowedBy() — 어떤 이벤트 유형이 다른 두 이벤트 유형 사이 어디에도 없게 하고 싶을 때.

경고: 시간 간격이 withIn()으로 정의되지 않은 경우 패턴 시퀀스는 notFollowedBy()로 끝날 수 없어요.

경고: NOT 패턴은 optional 패턴이 앞에 올 수 없어요.

// strict contiguity
Pattern<Event, ?> strict = start.next("middle").where(...);

// relaxed contiguity
Pattern<Event, ?> relaxed = start.followedBy("middle").where(...);

// non-deterministic relaxed contiguity
Pattern<Event, ?> nonDetermin = start.followedByAny("middle").where(...);

// NOT pattern with strict contiguity
Pattern<Event, ?> strictNot = start.notNext("not").where(...);

// NOT pattern with relaxed contiguity
Pattern<Event, ?> relaxedNot = start.notFollowedBy("not").where(...);

완화 연속성은 첫 번째로 성공하는 매칭 이벤트만 매치되는 것을 의미하는 반면, 비결정적 완화 연속성은 같은 시작점에 대해 여러 매치가 생성돼요. 예를 들어 패턴 "a b"가 이벤트 시퀀스 "a", "c", "b1", "b2"에 주어질 때 다음 결과를 보여줘요:

1. "a"와 "b" 사이 Strict Contiguity: {} (매치 없음), "a" 뒤의 "c"가 "a"를 버리게 해요.

2. "a"와 "b" 사이 Relaxed Contiguity: {a b1} — 완화 연속성은 "다음 매칭까지 비매칭 이벤트를 건너뛴다"로 보기 때문.

3. "a"와 "b" 사이 Non-Deterministic Relaxed Contiguity: {a b1}, {a b2} — 이것이 가장 일반적인 형태이기 때문.

패턴이 유효하기 위한 시간 제약 조건을 정의하는 것도 가능해요. 예를 들어 pattern.within() 메서드를 통해 패턴이 10초 안에 발생해야 한다고 정의할 수 있어요. 시간 패턴은 processing 및 event time 모두에서 지원돼요.

정보: 패턴 시퀀스는 하나의 시간 제약만 가질 수 있어요. 서로 다른 개별 패턴에 여러 제약이 정의되면 가장 작은 것이 적용돼요.

next.within(Duration.ofSeconds(10));

패턴 시퀀스는 시간 제약과 함께 notFollowedBy()로 끝날 수 있다는 점에 주의하세요. 예:

Pattern.<Event>begin("start")
    .next("middle")
    .where(SimpleCondition.of(value -> value.getName().equals("a")))
    .notFollowedBy("end")
    .where(SimpleCondition.of(value -> value.getName().equals("b")))
    .within(Duration.ofSeconds(10));

Contiguity within looping patterns

이전 section에서 논의한 것과 동일한 연속성 조건을 루핑 패턴 안에도 적용할 수 있어요. 연속성은 그런 패턴에 수용되는 요소들 사이에 적용돼요. 예를 들어, 패턴 시퀀스 "a b+ c"("a" 다음에 하나 이상의 "b"의 (비결정적 완화) 시퀀스, 그다음 "c")에 입력 "a", "b1", "d1", "b2", "d2", "b3" "c"가 주어지면 다음 결과가 나와요:

1. Strict Contiguity: {a b1 c}, {a b2 c}, {a b3 c} — 인접한 "b"가 없어요.

2. Relaxed Contiguity: {a b1 c}, {a b1 b2 c}, {a b1 b2 b3 c}, {a b2 c}, {a b2 b3 c}, {a b3 c} — "d"는 무시돼요.

3. Non-Deterministic Relaxed Contiguity: {a b1 c}, {a b1 b2 c}, {a b1 b3 c}, {a b1 b2 b3 c}, {a b2 c}, {a b2 b3 c}, {a b3 c} — "b" 사이의 연속성을 완화한 결과인 {a b1 b3 c}에 주목하세요.

루핑 패턴(예: oneOrMore()와 times())의 기본값은 완화 연속성이에요. 엄격 연속성을 원한다면 consecutive() 호출로 명시적으로 지정해야 하고, 비결정적 완화 연속성을 원한다면 allowCombinations() 호출을 사용할 수 있어요.

consecutive()

oneOrMore()와 times()와 함께 작동하며 매칭 이벤트 사이에 엄격한 연속성을 적용해요. 즉 비매칭 요소가 매치를 깨뜨려요(next()에서처럼). 적용하지 않으면 완화 연속성(followedBy()에서처럼)이 사용돼요.

예를 들어 패턴:

Pattern.<Event>begin("start")
    .where(SimpleCondition.of(value -> value.getName().equals("c")))
    .followedBy("middle")
    .where(SimpleCondition.of(value -> value.getName().equals("a")))
    .oneOrMore()
    .consecutive()
    .followedBy("end1")
    .where(SimpleCondition.of(value -> value.getName().equals("b")));

입력 시퀀스 C D A1 A2 A3 D A4 B에 대해 다음 매치를 생성할 텐데, consecutive 적용 시: {C A1 B}, {C A1 A2 B}, {C A1 A2 A3 B} consecutive 미적용 시: {C A1 B}, {C A1 A2 B}, {C A1 A2 A3 B}, {C A1 A2 A3 A4 B}.

allowCombinations()

oneOrMore()와 times()와 함께 작동하며 매칭 이벤트 사이에 비결정적 완화 연속성을 적용해요(followedByAny()에서처럼). 적용하지 않으면 완화 연속성(followedBy()에서처럼)이 사용돼요.

예를 들어 패턴:

Pattern.<Event>begin("start")
    .where(SimpleCondition.of(value -> value.getName().equals("c")))
    .followedBy("middle")
    .where(SimpleCondition.of(value -> value.getName().equals("a")))
    .oneOrMore()
    .allowCombinations()
    .followedBy("end1")
    .where(SimpleCondition.of(value -> value.getName().equals("b")));

입력 시퀀스 C D A1 A2 A3 D A4 B에 대해 다음 매치를 생성할 텐데, combinations 활성화 시: {C A1 B}, {C A1 A2 B}, {C A1 A3 B}, {C A1 A4 B}, {C A1 A2 A3 B}, {C A1 A2 A4 B}, {C A1 A3 A4 B}, {C A1 A2 A3 A4 B} combinations 미활성화 시: {C A1 B}, {C A1 A2 B}, {C A1 A2 A3 B}, {C A1 A2 A3 A4 B}.

Groups of patterns

begin, followedBy, followedByAny, next의 조건으로 패턴 시퀀스를 정의하는 것도 가능해요. 그 패턴 시퀀스는 논리적으로 매칭 조건으로 간주되고 GroupPattern이 반환되며, GroupPattern에 oneOrMore(), times(#ofTimes), times(#fromTimes, #toTimes), optional(), consecutive(), allowCombinations()을 적용할 수 있어요.

Pattern<Event, ?> start = Pattern.begin(
    Pattern.<Event>begin("start").where(...).followedBy("start_middle").where(...)
);

// strict contiguity
Pattern<Event, ?> strict = start.next(
    Pattern.<Event>begin("next_start").where(...).followedBy("next_middle").where(...)
).times(3);

// relaxed contiguity
Pattern<Event, ?> relaxed = start.followedBy(
    Pattern.<Event>begin("followedby_start").where(...).followedBy("followedby_middle").where(...)
).oneOrMore();

// non-deterministic relaxed contiguity
Pattern<Event, ?> nonDetermin = start.followedByAny(
    Pattern.<Event>begin("followedbyany_start").where(...).followedBy("followedbyany_middle").where(...)
).optional();

begin(#name)

시작 패턴을 정의해요.

Pattern<Event, ?> start = Pattern.<Event>begin("start");

begin(#pattern_sequence)

시작 패턴을 정의해요.

Pattern<Event, ?> start = Pattern.<Event>begin(
    Pattern.<Event>begin("start").where(...).followedBy("middle").where(...)
);

next(#name)

새 패턴을 추가해요. 매칭 이벤트가 이전 매칭 이벤트를 직접 이어야 해요(strict contiguity).

Pattern<Event, ?> next = start.next("middle");

next(#pattern_sequence)

새 패턴을 추가해요. 매칭 이벤트들의 시퀀스가 이전 매칭 이벤트를 직접 이어야 해요(strict contiguity).

Pattern<Event, ?> next = start.next(
    Pattern.<Event>begin("start").where(...).followedBy("middle").where(...)
);

followedBy(#name)

새 패턴을 추가해요. 매칭 이벤트와 이전 매칭 이벤트 사이에 다른 이벤트가 발생할 수 있어요(relaxed contiguity).

Pattern<Event, ?> followedBy = start.followedBy("middle");

followedBy(#pattern_sequence)

새 패턴을 추가해요. 매칭 이벤트와 이전 매칭 이벤트 사이에 다른 이벤트가 발생할 수 있어요(relaxed contiguity).

Pattern<Event, ?> followedBy = start.followedBy(
    Pattern.<Event>begin("start").where(...).followedBy("middle").where(...)
);

followedByAny(#name)

새 패턴을 추가해요. 매칭 이벤트와 이전 매칭 이벤트 사이에 다른 이벤트가 발생할 수 있고, 모든 대체 매칭 이벤트에 대해 대체 매치가 제시돼요(non-deterministic relaxed contiguity).

Pattern<Event, ?> followedByAny = start.followedByAny("middle");

followedByAny(#pattern_sequence)

새 패턴을 추가해요. 매칭 이벤트와 이전 매칭 이벤트 사이에 다른 이벤트가 발생할 수 있고, 모든 대체 매칭 이벤트에 대해 대체 매치가 제시돼요(non-deterministic relaxed contiguity).

Pattern<Event, ?> next = start.next(
    Pattern.<Event>begin("start").where(...).followedBy("middle").where(...)
);

notNext()

새 부정 패턴을 추가해요. 부분 매치가 버려지려면 매칭되는 (부정) 이벤트가 이전 매칭 이벤트를 직접 이어야 해요(strict contiguity).

Pattern<Event, ?> notNext = start.notNext("not");

notFollowedBy()

새 부정 패턴을 추가해요. 매칭 (부정) 이벤트와 이전 매칭 이벤트 사이에 다른 이벤트가 발생해도 부분 매칭 이벤트 시퀀스가 버려져요(relaxed contiguity).

Pattern<Event, ?> notFollowedBy = start.notFollowedBy("not");

within(time)

이벤트 시퀀스가 패턴을 매치할 최대 시간 간격을 정의해요. 완료되지 않은 이벤트 시퀀스가 이 시간을 초과하면 버려져요.

pattern.within(Duration.ofSeconds(10));

After Match Skip Strategy

주어진 패턴에서 같은 이벤트가 여러 매치에 할당될 수 있어요. 이벤트가 몇 개의 매치에 할당될지 제어하려면 AfterMatchSkipStrategy라는 스킵 전략을 지정해야 해요. 다섯 가지 유형의 스킵 전략이 있으며 다음과 같아요:

• NO_SKIP: 가능한 모든 매치가 생성돼요.
• SKIP_TO_NEXT: 매치가 시작된 것과 같은 이벤트로 시작한 모든 부분 매치를 버려요.
• SKIP_PAST_LAST_EVENT: 매치가 시작된 후 시작됐지만 매치가 끝나기 전에 끝난 모든 부분 매치를 버려요.
• SKIP_TO_FIRST: 매치가 시작된 후 시작됐지만 PatternName의 첫 이벤트가 발생하기 전에 끝난 모든 부분 매치를 버려요.
• SKIP_TO_LAST: 매치가 시작된 후 시작됐지만 PatternName의 마지막 이벤트가 발생하기 전에 끝난 모든 부분 매치를 버려요.

SKIP_TO_FIRST와 SKIP_TO_LAST 스킵 전략을 사용할 때는 유효한 PatternName도 지정해야 한다는 점에 주의하세요.

예를 들어 패턴 b+ c와 데이터 스트림 b1 b2 b3 c에 대해 네 가지 스킵 전략의 차이는 다음과 같아요:

Skip Strategy	Result	Description
NO_SKIP	b1 b2 b3 c b2 b3 c b3 c	매치 b1 b2 b3 c를 찾은 후 매치 과정이 어떤 결과도 버리지 않아요.
SKIP_TO_NEXT	b1 b2 b3 c b2 b3 c b3 c	매치 b1 b2 b3 c를 찾은 후 매치 과정이 어떤 결과도 버리지 않아요. b1에서 시작할 다른 매치가 없기 때문이에요.
SKIP_PAST_LAST_EVENT	b1 b2 b3 c	매치 b1 b2 b3 c를 찾은 후 매치 과정이 시작된 모든 부분 매치를 버려요.
SKIP_TO_FIRST[b]	b1 b2 b3 c b2 b3 c b3 c	매치 b1 b2 b3 c를 찾은 후 매치 과정이 b1 전에 시작된 모든 부분 매치를 버리려 시도하지만 그런 매치가 없어요. 따라서 아무것도 버려지지 않아요.
SKIP_TO_LAST[b]	b1 b2 b3 c b3 c	매치 b1 b2 b3 c를 찾은 후 매치 과정이 b3 전에 시작된 모든 부분 매치를 버리려 시도해요. b2 b3 c라는 매치가 하나 있어요.

NO_SKIP과 SKIP_TO_FIRST의 차이를 더 잘 보려면 다른 예를 봐도 좋아요: 패턴: (a | b | c) (b | c) c+.greedy d 그리고 시퀀스: a b c1 c2 c3 d 그러면 결과는:

Skip Strategy	Result	Description
NO_SKIP	a b c1 c2 c3 d b c1 c2 c3 d c1 c2 c3 d	매치 a b c1 c2 c3 d를 찾은 후 매치 과정이 어떤 결과도 버리지 않아요.
SKIP_TO_FIRST[c*]	a b c1 c2 c3 d c1 c2 c3 d	매치 a b c1 c2 c3 d를 찾은 후 매치 과정이 c1 전에 시작된 모든 부분 매치를 버려요. b c1 c2 c3 d라는 매치가 하나 있어요.

NO_SKIP과 SKIP_TO_NEXT의 차이를 더 잘 이해하려면 다음 예를 보세요: 패턴: a b+ 그리고 시퀀스: a b1 b2 b3 그러면 결과는:

Skip Strategy	Result	Description
NO_SKIP	a b1 a b1 b2 a b1 b2 b3	매치 a b1을 찾은 후 매치 과정이 어떤 결과도 버리지 않아요.
SKIP_TO_NEXT	a b1	매치 a b1을 찾은 후 매치 과정이 a에서 시작된 모든 부분 매치를 버려요. 이는 a b1 b2도 a b1 b2 b3도 생성될 수 없음을 의미해요.

사용할 스킵 전략을 지정하려면 AfterMatchSkipStrategy를 호출해 만드세요:

Function	Description
AfterMatchSkipStrategy.noSkip()	NO_SKIP 스킵 전략을 생성
AfterMatchSkipStrategy.skipToNext()	SKIP_TO_NEXT 스킵 전략을 생성
AfterMatchSkipStrategy.skipPastLastEvent()	SKIP_PAST_LAST_EVENT 스킵 전략을 생성
AfterMatchSkipStrategy.skipToFirst(patternName)	참조된 패턴 이름 patternName으로 SKIP_TO_FIRST 스킵 전략을 생성
AfterMatchSkipStrategy.skipToLast(patternName)	참조된 패턴 이름 patternName으로 SKIP_TO_LAST 스킵 전략을 생성

그런 다음 패턴에 스킵 전략을 적용하려면 다음을 호출하세요:

AfterMatchSkipStrategy skipStrategy = ...;
Pattern.begin("patternName", skipStrategy);

정보: SKIP_TO_FIRST/LAST의 경우 PatternName에 매핑된 이벤트가 없을 때 처리하는 두 가지 옵션이 있어요. 기본적으로 이 경우 NO_SKIP 전략이 사용돼요. 다른 옵션은 이런 상황에서 예외를 던지는 것이에요. 이 옵션은 다음으로 활성화할 수 있어요:

AfterMatchSkipStrategy.skipToFirst(patternName).throwExceptionOnMiss();

Detecting Patterns

찾고 있는 패턴 시퀀스를 지정한 후에는 그것을 입력 스트림에 적용해 잠재 매치를 감지할 차례예요. 패턴 시퀀스에 이벤트 스트림을 실행하려면 PatternStream을 만들어야 해요. 입력 스트림 input, 패턴 pattern, 그리고 EventTime의 경우 타임스탬프가 같거나 동시에 도착한 이벤트를 정렬하는 데 쓰는 선택적 비교자 comparator가 주어졌을 때 PatternStream은 다음 호출로 만듭니다:

DataStream<Event> input = ...;
Pattern<Event, ?> pattern = ...;
EventComparator<Event> comparator = ...; // optional

PatternStream<Event> patternStream = CEP.pattern(input, pattern, comparator);

입력 스트림은 사용 사례에 따라 keyed 또는 non-keyed일 수 있어요.

정보: non-keyed 스트림에 패턴을 적용하면 병렬도가 1인 잡이 생성돼요.

Selecting from Patterns

PatternStream을 얻으면 감지된 이벤트 시퀀스에 변환을 적용할 수 있어요. 권장하는 방법은 PatternProcessFunction을 사용하는 것이에요.

PatternProcessFunction에는 각 매칭 이벤트 시퀀스에 대해 호출되는 processMatch 메서드가 있어요. 그것은 Map<String, List<IN>> 형태의 매치를 받는데, 여기서 키는 패턴 시퀀스에 있는 각 패턴의 이름이고 값은 해당 패턴에 대해 수용된 모든 이벤트의 목록이에요(IN은 입력 요소의 유형). 주어진 패턴의 이벤트들은 타임스탬프로 정렬돼요. 각 패턴에 대해 수용된 이벤트 목록을 반환하는 이유는 루핑 패턴(예: oneToMany() 및 times())을 사용할 때 한 패턴에 대해 둘 이상의 이벤트가 수용될 수 있기 때문이에요.

class MyPatternProcessFunction<IN, OUT> extends PatternProcessFunction<IN, OUT> {
    @Override
    public void processMatch(Map<String, List<IN>> match, Context ctx, Collector<OUT> out) throws Exception;
        IN startEvent = match.get("start").get(0);
        IN endEvent = match.get("end").get(0);
        out.collect(OUT(startEvent, endEvent));
    }
}

PatternProcessFunction은 Context 객체에 접근을 제공해요. 덕분에 currentProcessingTime이나 현재 매치의 timestamp(매치에 할당된 마지막 요소의 타임스탬프) 같은 시간 관련 특성에 접근할 수 있어요. 자세한 내용은 Time context를 참고하세요. 이 컨텍스트를 통해 결과를 side-output으로도 보낼 수 있어요.

Handling Timed Out Partial Patterns

패턴에 within 키워드로 창 길이가 붙어 있으면 부분 이벤트 시퀀스가 창 길이를 초과해 버려질 수 있어요. 타임아웃된 부분 매치에 대응하려면 TimedOutPartialMatchHandler 인터페이스를 사용할 수 있어요. 이 인터페이스는 믹스인 방식으로 사용하게 돼요. 즉 PatternProcessFunction과 함께 이 인터페이스를 추가로 구현할 수 있음을 의미해요. TimedOutPartialMatchHandler는 타임아웃된 모든 부분 매치에 대해 호출되는 추가 processTimedOutMatch 메서드를 제공해요.

class MyPatternProcessFunction<IN, OUT> extends PatternProcessFunction<IN, OUT> implements TimedOutPartialMatchHandler<IN> {
    @Override
    public void processMatch(Map<String, List<IN>> match, Context ctx, Collector<OUT> out) throws Exception;
        ...
    }

    @Override
    public void processTimedOutMatch(Map<String, List<IN>> match, Context ctx) throws Exception;
        IN startEvent = match.get("start").get(0);
        ctx.output(outputTag, T(startEvent));
    }
}

정보: processTimedOutMatch는 기본 출력에 접근을 주지 않아요. 그래도 Context 객체를 통해 side-outputs으로 결과를 보낼 수 있어요.

Convenience API

앞서 언급한 PatternProcessFunction은 Flink 1.8에서 도입됐고 그 이후로 매치와 상호작용하는 권장 방식이에요. 여전히 select/flatSelect 같은 옛 스타일 API를 사용할 수 있는데, 내부적으로는 PatternProcessFunction으로 변환돼요.

PatternStream<Event> patternStream = CEP.pattern(input, pattern);

OutputTag<String> outputTag = new OutputTag<String>("side-output"){};

SingleOutputStreamOperator<ComplexEvent> flatResult = patternStream.flatSelect(
    outputTag,
    new PatternFlatTimeoutFunction<Event, TimeoutEvent>() {
        public void timeout(
                Map<String, List<Event>> pattern,
                long timeoutTimestamp,
                Collector<TimeoutEvent> out) throws Exception {
            out.collect(new TimeoutEvent());
        }
    },
    new PatternFlatSelectFunction<Event, ComplexEvent>() {
        public void flatSelect(Map<String, List<IN>> pattern, Collector<OUT> out) throws Exception {
            out.collect(new ComplexEvent());
        }
    }
);

DataStream<TimeoutEvent> timeoutFlatResult = flatResult.getSideOutput(outputTag);

Time in CEP library

Handling Lateness in Event Time

CEP에서는 요소가 처리되는 순서가 중요해요. 이벤트 타임에서 요소가 올바른 순서로 처리되도록 보장하기 위해 들어오는 요소는 처음에 타임스탬프 기준 오름차순으로 정렬된 버퍼에 들어가고, 워터마크가 도착하면 이 버퍼에서 워터마크보다 작은 타임스탬프를 가진 모든 요소가 처리돼요. 이는 워터마크 사이의 요소들이 이벤트 타임 순서대로 처리된다는 것을 의미해요.

정보: 라이브러리는 이벤트 타임에서 작업할 때 워터마크의 정확성을 가정해요.

워터마크를 넘어서는 요소가 이벤트 타임 순서로 처리되도록 보장하기 위해 Flink의 CEP 라이브러리는 워터마크의 정확성을 가정하고, 타임스탬프가 마지막으로 본 워터마크보다 작은 요소를 늦은(late) 것으로 간주해요. 늦은 요소는 더 이상 처리되지 않아요. 또한 마지막으로 본 워터마크 이후에 도착한 늦은 요소를 수집하기 위한 sideOutput 태그를 지정할 수 있는데, 다음과 같이 사용해요.

PatternStream<Event> patternStream = CEP.pattern(input, pattern);

OutputTag<String> lateDataOutputTag = new OutputTag<String>("late-data"){};

SingleOutputStreamOperator<ComplexEvent> result = patternStream
    .sideOutputLateData(lateDataOutputTag)
    .select(
        new PatternSelectFunction<Event, ComplexEvent>() {...}
    );

DataStream<String> lateData = result.getSideOutput(lateDataOutputTag);

Time context

PatternProcessFunction과 IterativeCondition 모두에서 사용자는 TimeContext를 구현하는 컨텍스트에 접근할 수 있어요:

/**
 * Enables access to time related characteristics such as current processing time or timestamp of
 * currently processed element. Used in {@link PatternProcessFunction} and
 * {@link org.apache.flink.cep.pattern.conditions.IterativeCondition}
 */
@PublicEvolving
public interface TimeContext {

	/**
	 * Timestamp of the element currently being processed.
	 *
	 * <p>In case of {@link org.apache.flink.streaming.api.TimeCharacteristic#ProcessingTime} this
	 * will be set to the time when event entered the cep operator.
	 */
	long timestamp();

	/** Returns the current processing time. */
	long currentProcessingTime();
}

이 컨텍스트는 사용자에게 처리되는 이벤트(IterativeCondition의 경우 들어오는 레코드, PatternProcessFunction의 경우 매치)의 시간 특성에 접근을 주어요. TimeContext#currentProcessingTime 호출은 항상 현재 처리 시간 값을 주며, 이 호출이 예를 들어 System.currentTimeMillis()를 호출하는 것보다 선호되어야 해요.

TimeContext#timestamp()의 경우 반환 값은 EventTime의 경우 할당된 타임스탬프와 같아요. ProcessingTime에서는 해당 이벤트가 cep 연산자에 들어간 시점(PatternProcessFunction의 경우 매치가 생성된 시점)과 같아요. 이는 여러 호출에서 값이 일관된다는 것을 의미해요.

Optional Configuration

Flink CEP SharedBuffer의 캐시 용량을 구성하는 옵션이에요. CEP 처리 속도를 높이고 순수 메모리에서 캐시 요소의 수를 제한할 수 있어요.

정보: state.backend.type이 rocksdb로 설정된 경우에만 메모리 사용 제한에 효과적이에요. 캐시 수를 초과하는 요소를 메모리 상태 저장소 대신 rocksdb 상태 저장소로 옮기기 때문이에요.

state.backend.type이 rocksdb로 설정된 경우 메모리 제한에 도움이 되는 구성 항목이에요. 반대로 state.backend.type이 rocksdb가 아니면 캐시는 성능 저하를 일으켜요. 옛 Map으로 구현된 캐시와 비교해 상태 부분에는 새 guava-cache에서 스왑된 요소가 더 많아져 상태의 copy on write가 더 무거워질 수 있어요.

Examples

다음 예는 keyed 데이터 스트림 Events에서 패턴 start, middle(name = "error") -> end(name = "critical")을 감지해요. 이벤트는 id로 키가 정해지고 유효한 패턴은 10초 안에 발생해야 해요. 전체 처리는 이벤트 타임으로 수행돼요.

StreamExecutionEnvironment env = ...;

DataStream<Event> input = ...;

DataStream<Event> partitionedInput = input.keyBy(new KeySelector<Event, Integer>() {
	@Override
	public Integer getKey(Event value) throws Exception {
		return value.getId();
	}
});

Pattern<Event, ?> pattern = Pattern.<Event>begin("start")
    .next("middle")
    .where(SimpleCondition.of(value -> value.getName().equals("error")))
    .followedBy("end")
    .where(SimpleCondition.of(value -> value.getName().equals("critical")))
    .within(Duration.ofSeconds(10));

PatternStream<Event> patternStream = CEP.pattern(partitionedInput, pattern);

DataStream<Alert> alerts = patternStream.select(new PatternSelectFunction<Event, Alert>() {
	@Override
	public Alert select(Map<String, List<Event>> pattern) throws Exception {
		return createAlert(pattern);
	}
});

Migrating from an older Flink version(pre 1.5)

Migrating from Flink <= 1.5

Flink 1.13에서 Flink <= 1.5와의 직접 savepoint 역호환성을 제거했어요. 이전 버전에서 가져온 savepoint로 복원하려면 먼저 더 새로운 버전(1.6-1.12)으로 마이그레이션하고 savepoint를 만든 다음 그 savepoint로 Flink >= 1.13에서 복원하세요.

더 알아보기 (Learn more)

• Generating Watermarks
• Side Outputs
• Process Function
• Time (concepts)