포스코홀딩스의 Managed Apache Flink를 활용한 효율적인 다수 CCTV 이벤트 처리 사례

포스코홀딩스는 한국의 대표적인 글로벌 철강 기업 포스코를 중심으로 이차전지 소재, 건설 등 다양한 사업 포트폴리오를 보유한 그룹 지주사입니다. 그룹 전체의 지속 가능한 성장을 추구하며, 특히 안전을 최우선 가치로 두고 있습니다. 사업장 내 안전사고 예방을 위해 끊임없이 노력하고 있습니다. 최근 AI 기술의 급격한 발전과 함께, 포스코홀딩스는 AI를 활용한 실시간 상황감지 시스템 구축에 많은 관심을 기울이고 있습니다.

현재 포항과 광양제철소에는 고정식, 이동식, 바디캠 등 약 3만대 이상의 CCTV를 운영하고 있고 통합관제센터는 작업자와 설비 안전관리를 위해 CCTV를 통해 작업 현장을 실시간 모니터링하고 있습니다. 사고 발생 가능성을 예측하여 예방하고자 기존 CCTV 시스템에 AI 기술을 접목, 스마트 안전 시스템으로의 전환을 추진하고 있습니다.

이에 포스코홀딩스는 기존 AI 모델과 함께 실시간 데이터 스트리밍 서비스인 HAQM Kinesis와 실시간 데이터 분석 및 복합 이벤트 처리(CEP, Complex Event Processing)를 위해 HAQM Managed Service for Apache Flink를 사용하였습니다. Kinesis를 통해 실시간으로 수집되는 다양한 CCTV 영상에 대한 AI 추론 데이터를 Managed Apache Flink에서 분석하고, 이벤트 기반의 위험 상황 판단 로직과 AI 모델의 예측 결과를 결합하여 더욱 정확하고 신속한 위험 감지를 구현하고자 했습니다.

AWS 팀과 협력하여 Kinesis와 Managed Apache Flink 서비스를 성공적으로 플랫폼에 도입했습니다. 이 블로그에서는 포스코홀딩스의 과제 적용 사례와 함께 구현한 솔루션을 공유합니다. 산업재해 예방시스템 구축을 모색하는 많은 기업들에게 도움이 되었으면 합니다.

HAQM Kinesis, HAQM Managed Apache Flink

Kinesis는 AWS에서 제공하는 실시간 데이터 스트리밍 플랫폼입니다. 대규모의 실시간 데이터 스트림을 안정적이고 확장 가능하게 수집, 처리, 저장할 수 있도록 설계되었습니다.

Managed Apache Flink는 AWS에서 제공하는 완전 관리형 Apache Flink서비스입니다. Apache Flink는 대규모 데이터 스트림을 처리하기 위한 오픈 소스 분산 스트림 처리 엔진입니다. AWS는 Flink 클러스터의 설정, 관리, 확장 등의 운영 부담을 덜어주고, 사용자는 데이터 처리 로직 개발에 집중할 수 있도록 지원합니다.

또한 Apache Flink CEP (Complex event processing)는 Flink의 강력한 기능 중 하나로, 실시간으로 복잡한 이벤트 패턴을 감지하는 데 특화된 라이브러리입니다. CEP는 단순한 이벤트 스트림에서 의미 있는 패턴이나 시퀀스를 식별하여 즉각적인 대응을 가능하게 합니다.

Solution overview

포스코홀딩스는 안전 분야에 AI 기술을 접목하며 그룹사 안전을 강화하고자 기존에 운영 중이던 일반 CCTV의 스마트화를 진행하고 있습니다.

그림1은 포스코홀딩스가 개발한 실시간 상황감지 플랫폼의 일부 핵심 부분에 대한 아키텍처를 보여줍니다.

그림 1 포스코홀딩스의 자연어 기반 상황감지 플랫폼 아키텍쳐

포스코홀딩스의 실시간 상황감지 플랫폼은 On-premise에서 운영 중인 다채널 CCTV 영상으로부터 AI 모델 추론 결과를 Kinesis로 수집하고 Managed Apache Flink와 Flink CEP를 사용하여 실시간으로 복합 이벤트 패턴을 감지할 수 있습니다.

특히, 안전 애플리케이션은 작업 상황과 장소에 따라 감지 로직의 수정과 업데이트가 필요할 수 있습니다. 이러한 유연성을 확보하기 위해 AWS 서비스를 활용하여 애플리케이션을 손쉽게 재배포할 수 있는 파이프라인을 구성하였습니다.

또한, Managed Apache Flink 서비스를 통해 복수의 스트리밍 애플리케이션을 배포할 수 있어, 각 현장에 필요한 로직을 개별적으로 설정할 수 있도록 하였습니다. 따라서 다양한 상황에 대응하는 감지 로직을 구현하는 것이 중요했습니다.

Flink 스트리밍 애플리케이션 구현은Flink CEP 라이브러리를 사용할 수 있는 Java를 사용하기로 결정했습니다. 로컬에서 애플리케이션을 개발하고 테스트한 뒤에 Jar로 빌드하여 Managed Apache Flink의 스트리밍 애플리케이션에 배포할 수 있습니다.

Managed Apache Flink에 배포된 스트리밍 애플리케이션에는 간단한 하위 이벤트부터 이벤트 간 관계를 통해 도출되는 상위 이벤트까지 검출할 수 있는 로직을 포함하고 있습니다. 장소나 상황에 따라 Flink App의 로직을 변경하여 신규, 재배포가 가능했습니다.

한편, 이벤트 감지가 실시간으로 이루어져야 했기 때문에 지연이 발생하지 않도록 하는 것이 중요했습니다. 또한  감지된 이벤트 중에서 특정 이벤트의 발생 여부를 확인하는 효율적인 방법에 대한 고민이 있었으며, 이 문제를 해결하기 위한 적절한 솔루션을 찾았습니다.

Problem

• 이벤트 필터링을 위한 비용 효율적인 파이프라인의 부재
• 다양한 이벤트 감지 시나리오에서의 Flink CEP 로직의 복잡성
• Managed Apache Flink의 장기 운영 시, 누적되는 처리 지연 문제

Solution

이벤트 필터링을 위한 Managed Apache Flink, CloudWatch

Managed Apache Flink에서 생성된 감지 결과 이벤트를 Kinesis로 Sink할 때, 특정 이벤트 발생 시 이에 따른 액션을 취해야 하는 시나리오가 존재합니다.

이 경우, 특정 이벤트가 발생하는지 주기적으로 확인하기 위해 Kinesis 스트림을 직접 구독하여 이벤트를 모니터링하는 방법을 고려할 수 있습니다. 이를 위해서는 각 샤드의 추가, 구독 애플리케이션들의 Read 데이터 양과 속도 조정이 필요하며, 향상된 팬아웃 (Enhanced Fan Out, EFO)를 활용하면 보다 안정적인 레코드 구독 및 분산 처리가 가능해집니다.

그러나, EFO 사용 시 추가 비용이 발생하는 단점이 있습니다. 또한 HAQM EventBridge나 HAQM Data Firehose를 이용하면 AWS 서비스 간 이벤트 발생에 따른 후속 처리를 연동하기 쉽지만, 탐지하고자 하는 이벤트 종류가 수시로 변경되고 결과 이벤트를 지속적으로 필터링해야 하는 경우 서비스 구성이 복잡해질 수 있습니다.

따라서, 이러한 비용 부담과 운영 복잡성을 해소하기 위해, HAQM CloudWatch와 구독 필터(Subscription Filter)를 활용하는 방법을 선택하였습니다.

그림 2 이벤트 필터링 파이프라인

그림2는 사용자가 감지 로직을 생성하거나 업데이트하고 Managed Apache Flink 스트리밍 애플리케이션을 자동으로 배포하며, 구동 중인 스트리밍 애플리케이션에서 감지하고 싶은 특정 이벤트를 선별적으로 HAQM RDS에 저장하는 과정을 보여줍니다.

각 스트리밍 애플리케이션에서는 Apache Log4j를 활용해 로그를 HAQM CloudWatch로 전송하며, 이를 통해 기록된 로그를 모니터링할 수 있도록 설계하였습니다.

그림 3 이벤트 데이터 흐름과 이벤트 필터 변경

그림3은 사용자가 Notification Service와 연동하고 싶은 이벤트를 선택하여 필터링하는 과정을 보여줍니다. Managed Apache Flink에서 나온 모든 이벤트를 로그 그룹으로 전달하고 구독 필터를 사용해서 선택한 이벤트가 감지되면 HAQM SNS, HAQM SES 등의 알림 서비스를 실행할 수 있습니다. 또한 구독 필터에서 패턴 필터링을 쉽게 변경할 수 있기 때문에 변경된 이벤트에 대한 구독을 빠르게 적용할 수 있습니다.

그림 4 구독 필터 적용 결과

그림4는 구독 필터 변경이 적용된 결과를 보여줍니다. 이를 통해 스트리밍 애플리케이션에서 직접 Kinesis 스트림을 구독하여 발생할 수 있는 오류를 회피하고, 안정적으로 특정 이벤트 감지를 수행할 수 있습니다.

따라서, Managed Apache Flink에서 발생한 이벤트가 Kinesis로 Sink된 후 특정 이벤트를 기반으로 액션을 수행해야 하는 상황에서, 직접적인 스트림 구독으로 인한 위험을 줄이고 추가 비용 부담 없이 CloudWatch와 구독 필터를 활용하여 안정적인 후처리 시스템을 구축할 수 있음을 확인할 수 있습니다.

다양한 이벤트 감지 시나리오에서의 Flink CEP 로직을 효과적으로 구현

Flink CEP는 스트림 데이터 내에서 복잡한 이벤트 패턴을 정의하고 탐지하는 데 특화된 라이브러리입니다. CEP를 사용하면 비교적 간단한 형태의 패턴 로직을 선언적으로 표현하여, 실시간 스트림 처리 중에 일정한 순서나 조건을 만족하는 이벤트 집합을 감지할 수 있습니다.특정 위험 상황에 해당하는 이벤트 시퀀스나 조건을 CEP 패턴으로 정의해, 그 패턴에 매칭되는 경우에만 경보를 발생시키거나 추가 처리를 할 수 있습니다.

다음은 CEP를 활용하여 이벤트 감지 시나리오를 효과적으로 구현하는 방안에 대한 몇 가지 핵심 포인트입니다.

1) KeyedProcessFunction
KeyedProcessFunction함수는 개별 key 단위로 데이터를 처리할 수 있으며, 이벤트 기반 처리 및 타이머(timer) 기능을 함께 사용할 수 있게 해줍니다. 실제 이 함수가 유용하게 쓰이는 경우에 대해 설명하겠습니다.

예를 들어, Vision AI 인식 결과로부터 추출된 여러 사람 혹은 객체를 그들의 고유 식별자(ID)에 따라 그룹화하여, 동일한 ID를 가진 객체들끼리 집계 및 개별 처리가 가능합니다. 동일 객체에 대한 움직임 여부 판단, 특정 객체와의 접근 판단 등 후속 처리에 사용할 수 있습니다.

//입력스트림과 패턴 연결
PatternStream<VisionObjectDetectionData> patternStream = CEP.pattern(inputStream, pattern);

//매칭 패턴 처리
DataStream< VisionObjectDetectionData > resultProcessStream = patternStream
                .inEventTime()
                .process(new PatternProcessFunction<>() {
                    @Override
                    public void processMatch() throws Exception {
                        out.collect(MatchedEvent);
                    }
                });

//매칭 패턴 후처리
DataStream<String> windowedPreprocessedStream = resultProcessStream
        // 객체 ID 기준
    .keyBy(event -> event.getId())
    .process(new KeyedProcessFunction<Integer, VisionObjectDetectionData, String>() {        
        // key 단위로 상태를 저장하기 위한 ValueState 선언
        private transient ValueState<Integer> durationState;

        @Override
        public void open(Configuration parameters) throws Exception {
            // StateDescriptor를 정의, 초기화
            ValueStateDescriptor<Integer> durationDescriptor = new ValueStateDescriptor<>(
                    "durationDescriptor",
                    int.class
            );
            durationState = getRuntimeContext().getState(durationDescriptor);
        }

        @Override
        public void processElement(VisionObjectDetectionData value, KeyedProcessFunction<Integer, VisionObjectDetectionData, String>.Context ctx, Collector<String> out) throws Exception {
            // 패턴 결과에 따라 새로운 이벤트 생성
            NewEventData alertData = NewEventData.builder()
                                    .build();
            //생성된 이벤트 출력
            out.collect(alertData.toString());

            // 타이머 설정
            ctx.timerService().registerEventTimeTimer(ctx.timestamp() + Time.seconds(5).toMilliseconds());
        }
        @Override
        public void onTimer(long timestamp, OnTimerContext ctx, Collector<String> out) {
            // 타이머가 실행될 때 durationState 상태를 리셋
            durationState.clear();
        }
    }); 

2) Windows Operator
패턴 매칭 후 새로운 이벤트를 발생시키는 로직은 단순히 매치가 될 때마다 이벤트를 무분별하게 생성할 경우, 중복 이벤트 발생이나 불필요한 알람을 야기할 수 있습니다. 이를 방지하기 위해 후처리 로직을 정교하게 설계해야 하고 이 때 Windows를 활용할 수 있습니다.

예를 들어, 사람이 지게차와 근접한 경우에는 지속적으로 일정 주기마다 패턴 매칭 이벤트를 생성해야 합니다. 이러한 경우, 그림과 같이 Tumbling Windows를 사용하여 이벤트 발생 주기를 컨트롤 할 수 있습니다.

DataStream<String> windowedPreprocessedStream = resultProcessPatternStream
            //매칭 패턴들을 5초 간격으로 모아서 새로운 이벤트를 발생
            .windowAll(TumblingEventTimeWindows.of(Time.seconds(5)))
                        // Adopt last data
            .reduce((e1, e2) -> e2)
                        // 후처리
            .process(); 

하지만 Tumbling Windows는 즉각적인 이벤트 감지 결과를 내보내야하는 시나리오에 적합하지 않을 수 있습니다. 패턴이 매칭되자마자 새로운 이벤트를 발생시켜야 하고 동시에 자주 발생되지 않도록 일정 시간 동안은 간격을 두게 하기 위해서는 아래와 같이 코드를 작성할 수 있습니다.

DataStream<String> processedStream = resultProcessStream
    .keyBy(event -> event.getId())
    .process(new KeyedProcessFunction<Integer, VisionObjectDetectionData, String>() {
        // ValueState 정의
        private transient ValueState<Boolean> processedState;

        @Override
        public void open(Configuration parameters) throws Exception {
            // StateDescriptor를 정의하고 초기화합니다.                        
        }

        @Override
        public void processElement(VisionObjectDetectionData value, KeyedProcessFunction<Integer, VisionObjectDetectionData, String>.Context ctx, Collector<String> out) throws Exception {
            // ValueState업데이트
            Boolean processedState = processedState.value();
            Long prevEventTimeStamp = prevEventTimeStampState.value();

            // 패턴 매칭 시점에 새로운 이벤트를 생성, 후속 이벤트는 일정 시간 이후에 생성하도록 처리
            if (processedState == null || ! processedState) {
processedState.update(true);
                prevEventTimeStampState.update(value.getEventTimestamp());
                ctx.timerService().registerEventTimeTimer(ctx.timestamp() + Time.seconds(5).toMilliseconds());
                AlertEventData alertData = AlertEventData.builder()
                        .build();
                out.collect(alertData.toString());
            }
        }
        @Override
        public void onTimer(long timestamp, OnTimerContext ctx, Collector<String> out) {
            // 타이머가 실행될 때 processedState 상태를 리셋하여 다음 첫 이벤트도 처리할 수 있도록 합니다.
            processedState.clear();
        }
    });

마지막으로 Session Windows를 활용한 설비 안전 점검 시나리오를 설명하겠습니다.

예를 들어, 작업자가 지정된 영역에 1시간 간격으로 위치해야 하며, 시간을 초과한 경우 알람을 발생해야 합니다. Session Windows를 적용하면, 작업자가 영역에 처음 들어올 때 해당 세션이 시작됩니다. 이후 1시간 이내에 작업자가 계속 감지되면 세션은 자동으로 연장되지만, 1시간 이상 추가 이벤트가 감지되지 않을 경우 자동으로 종료되며, 이 종료 시점에 알람을 발생시킬 수 있습니다.

DataStream<String> windowedPreprocessedStream = resultProcessPatternStream
            .keyBy(event -> event.getId())
            // Session Windows 설정: 
            .window(EventTimeSessionWindows.withGap(Time.seconds(3600)))
                      // 후처리
            .process();

위와 같이, 시나리오에 적합한 Windows Operator를 사용해야 불필요한 중복 이벤트 발생을 방지할 수 있으며 후속 조치를 정확하게 수행할 수 있게 됩니다.

Managed Apache Flink의 장기 운영 시, 누적되는 처리 지연 문제 해결

그림 5와 같이 CEP 기능이 포함된 Flink 애플리케이션을 Managed Apache Flink에서 장기 운영 하는 경우, CEP operator에서 사용률(Busy Rate)이 점점 증가하고 이로 인해 역압(Backpressure)가 발생할 우려가 있습니다. 이는 점점 처리 지연이 누적되어 감지 시점과 실제 발생 시점 간의 차이가 발생합니다. 이에 대한 원인은 매우 다양할 수 있어 단계적으로 해결 방안을 살펴보았습니다.

그림 5 Flink dashboard: 개별 operator에 대한 상태 확인

문제 1) 패턴 매칭 조건으로 인한 과도한 CEP 연산

그림 6은 실제 AI 모델의 추론 결과 중 일부 데이터입니다. 간단한 예시로, 사람의 출입이 잦은 곳에서 사람이 지게차와 충돌 위험 거리 이내에 있는 경우를 감지하는 패턴을 구현하고자 합니다. 같이 있다는 조건은 100ms 이내에 두 객체 이벤트를 수신한 경우로 가정하여 들어보겠습니다.

그림 6 AI 모델의 CCTV 비디오 추론 결과 데이터 예시

Pattern<Event, ?> detectForkliftWithPerson = Pattern.<Event>begin("start")
    .where(new SimpleCondition<Event>() {
        @Override
        public boolean filter(Event event) throws Exception {
            return event.getCls().equals("person");
        }
    })
    .followedBy("end")
    .where(new SimpleCondition<Event>() {
        @Override
        public boolean filter(Event event) throws Exception {
            return event.getCls().equals("forklift");
        }
    }).within(Time.milliseconds(100));

<구현 1>패턴의 첫 번째 조건이 “사람(person)”일 때

Pattern<Event, ?> detectForkliftWithPerson = Pattern.<Event>begin("start")
    .where(new SimpleCondition<Event>() {
        @Override
        public boolean filter(Event event) throws Exception {
            return event.getCls().equals("forklift");
        }
    })
    .followedBy("end")
    .where(new SimpleCondition<Event>() {
        @Override
        public boolean filter(Event event) throws Exception {
            return event.getCls().equals("person");
        }
    }).within(Time.milliseconds(100));

<구현 2> 패턴의 첫 번째 조건이 “지게차(forklift)”일 때

해결 방안: 효율적인 패턴 조건 최적화와 상태 관리하여 CEP 지연을 방지

Flink CEP는 패턴 매칭 조건을 만족하거나 만료된 경우, 내부에서 상태를 정리합니다.
구현1과 구현2는 동일한 결과를 보입니다. 물론 선행, 후행 조건이 다르지만 빠른 속도로 추론 결과가 데이터 스트림으로 들어오는 상황에서 미세한 시간 차이만 존재할 것입니다.

그러므로, 앞서 사람이 잦은 곳이라는 조건에서는 첫번째 매칭 조건을 지게차로 가져가는 코드가 더 효율적입니다. 매칭 조건을 기다리는 상태가 더 적기 때문입니다.

이와 같이, 단순한 순서 변경이나 시작 조건을 타이트하게 하여 부분 일치(Partial Matches)를 줄이므로 CEP Operator의 Busy rate를 낮출 수 있습니다. 또한, Pattern API사용 시  oneOrMore()나 times() 같은 Looping Patterns 사용을 주의하고 적절한 Skip Strategies를 사용하는 것도 개선에 도움이 될 수 있습니다.

문제2) State Backend 설정으로 인한 지연

Managed Apache Flink로 배포한 애플리케이션을 장시간 구동 시에 지연 문제가 발생하였으며, 패턴 수정, KPU 리소스 확대, 병렬 처리 설정 변경, 기타 리소스 옵션 변경을 통하여 이벤트 처리 지연을 해결하기 위해 많은 시도를 해보았지만 큰 효과는 없었습니다.

로컬 테스트 환경에서 State Backends Configuration을 MemoryStateBackend(Hashmap)로 변경하여 장시간 구동 시 에는 지연 문제가 발생하는 현상이 지속되지 않았습니다. State Backend는 스트림 처리 작업에서 발생하는 상태(State)를 어디에 어떻게 저장할지를 결정하는 컴포넌트입니다.

각각의 State Backend는 상태 저장 방식과 내구성, 성능, 확장성 측면에서 특성이 다르기 때문에 상황에 맞는 선택이 필요합니다. 또한, Flink CEP는 복잡한 패턴을 감지하기 위해 내부적으로 State를 유지하기 때문에 State Backend와 긴밀한 관계가 있음을 알게 되었습니다.

Managed Apache Flink에서는 대규모 상태 처리와 관리에 특화된 RocksDBStateBackend로 세팅이 되어 있습니다. 로컬 테스트 환경 설정을 RocksDBStateBackend로 변경하여 문제를 재현해 보았습니다. 그림 7은 VisualVM 결과이며, CPU 사용량을 보니 GlobalCepOperator 하위에 RocksDB에 대한 리소스 소모가 많은 것을 확인했습니다.

그림 7 CPU 리소스: GlobalCepOpertor 연산 비중에서 RocksDB 관련 연산량이 많음

해결방안: Managed Apache Flink의 State Backends Configration 변경

AWS Support 팀의 기술 지원을 받아 그림8 처럼 State Backend를 MemoryStateBackend(Hashmap)으로 변경하여 처리 시간 지연 문제를 해결할 수 있었습니다.

그림 8 The setting of the state backend is MemoryStateBackend (hashmap)

여러 CEP 패턴이 병렬로 돌아가고 패턴 매칭을 수행하며 이벤트를 발생시키는 애플리케이션 특징으로 상태의 크기가 크지 않고, 패턴 매칭 로직이 빈번하게 실행되는 애플리케이션에서는 State Backend를 매우 빠르게 접근하고 업데이트하는 것이 필요합니다. 그렇기 때문에 MemoryStateBackend를 사용하였습니다.

물론 State Backend를 MemoryStateBackend로 변경하는 것이 항상 정답은 아닐 수 있습니다. 이와 달리 대량의 데이터를 집계하거나 분석하는 경우에는 상태가 매우 커질 수 있기 때문에 메모리 내에서 관리하기보다 RocksDB와 같은 디스크 기반의 State Backend가 더 유리할 수 있습니다.

Conclusion

이 게시물에서는 포스코홀딩스의 실시간 상황감지 플랫폼이 On-premise 망에서 운영 중인 다채널 CCTV 영상 데이터를 활용하여 어떻게 실시간 복합 이벤트를 감지하는지 그 과정과 해결 방안을 소개했습니다.

이 플랫폼은 AI 모델 추론 결과를 Kinesis Data Stream으로 수집하고, Managed Apache Flink와 Flink CEP를 통해 다양한 상황에 따른 이벤트 패턴을 실시간으로 탐지할 수 있도록 구축되었습니다. AWS 팀과 협력하여 시나리오 요구사항에 맞춰 작업 상황과 장소별로 감지 로직을 유연하게 수정하고 업데이트할 수 있는 파이프라인을 구현했습니다.

특히, Managed Apache Flink 서비스를 활용하여 각 현장마다 독자적으로 배포 가능한 스트리밍 애플리케이션을 구성함으로써, 현장별 맞춤형 로직 적용 및 재배포가 가능하도록 하였습니다. 복잡한 이벤트 감지 요구사항을 효과적으로 처리하여 시스템의 전반적인 효율성과 안정성을 향상시켰습니다. 이를 통해 실시간 대응 능력을 강화하고, 운영 효율성을 높이는 효과를 얻을 수 있었습니다.

시스템의 확장성과 안정성을 더욱 높이기 위해 자동화된 배포 파이프라인과 모니터링 기능을 보강하여, 운영 효율을 극대화하는 방향으로 연구와 개발을 진행할 계획입니다. 아울러, 실시간 상황 감지 플랫폼의 활용도를 높이기 위해 CCTV 영상 외에도 추가적인 센서 데이터와 IoT 디바이스를 연동하여 더욱 정밀하고 신뢰도 높은 상황 분석 결과를 도출하고 이를 통해 실시간 의사 결정을 지원할 수 있도록 발전시킬 계획입니다.