Flink SQL 도입기

 (hyperconnect.github.io)
3P by GN⁺ | ★ favorite | 댓글 1개
  • Hyperconnect Azar Matching Dev Team은 CPU 96개를 쓰던 모놀리식 Flink 레거시 앱을 쪼개기 위해, 애플리케이션 코드 대신 SQL로 스트리밍 처리를 구현하는 방식을 선택함
  • 여러 Flink App으로 나누면 격리는 좋아지지만 운영 부담이 커져, 팀의 Flink 내부 이해도를 활용할 수 있는 Flink SQL이 생산성과 운영 효율성 면에서 더 적합하다고 판단함
  • Flink SQL은 Checkpoint/Savepoint, JobManager HA, TaskManager 재분배, 윈도우·조인·이벤트 타임·워터마크, UDF와 커스텀 커넥터를 지원해 ksqlDB와 Spark Structured Streaming보다 팀 상황에 맞았음
  • 운영 환경은 Kubernetes 위의 Session mode Flink Cluster로 구성했고, Flink SQL Gateway API와 GitHub Actions를 이용해 GitOps 방식으로 쿼리 배포와 Job 중단을 처리함
  • 약 1년간 안정적으로 운영되며 확대 중이지만, 쿼리 재배포와 클러스터 설정 변경은 여전히 불편해 GitOps Controller 패턴 기반 개선을 계획함

무거운 레거시 스트리밍 앱을 대체한 배경

  • Azar Matching Dev Team은 여러 Flink 기반 앱을 관리하고 있었고, 그중 CPU를 96개 사용하는 무거운 레거시 앱이 있었음
  • 이 앱은 여러 매치 이벤트 조인, 조건부 이벤트 발행, Redis 플래그 저장 같은 기능을 한곳에 모은 모놀리식 구조였음
  • 전사 인프라 작업으로 실행 노드를 변경한 뒤 앱이 정상 동작하지 않았고, 단순 튜닝만으로는 빠르게 해결하기 어려웠음
  • 중요한 이벤트 조인 기능은 별도 프로젝트의 새 Flink 앱에 이미 구현돼 있었기 때문에, 이벤트 조인 이후의 조건부 이벤트 발행과 로직 수행 부분을 대체하는 방식이 필요했음

대체 방식 비교

  • 하나의 Flink App으로 구현하면 관리 대상은 적지만, 다시 거대한 앱이 될 가능성이 크고 한 부분의 실패가 다른 기능에 영향을 줄 수 있음
  • 여러 Flink App으로 나누면 각 앱을 독립적으로 관리할 수 있으나, 앱 수가 늘수록 클러스터·리소스·배포 부담도 함께 커짐
  • Flink SQL은 쿼리로 로직을 정의해 빠르게 개발하고 하나의 클러스터만 관리할 수 있지만, 복잡한 로직을 SQL만으로 표현하기 어렵고 클러스터 운영 경험이 필요함
  • 팀은 Flink 내부 구현에 대한 이해도가 높아진 상태였고, Flink SQL이 생산성과 운영 효율성에서 이점이 있다고 판단함

Flink SQL을 선택한 이유

  • Flink SQL은 애플리케이션 코드를 직접 작성하지 않고도 SQL로 이벤트 스트리밍 처리 앱을 구현할 수 있음
  • 고가용성(HA) 측면에서 Flink는 상태 저장 처리를 지원하고, Checkpoint와 Savepoint로 작업 상태를 주기적이거나 원하는 시점에 저장·복구할 수 있음
    • JobManager는 leader-standby 형태의 HA 모드로 구성 가능함
    • TaskManager 일부가 실패하면 Job 재시도 전략에 따라 실패한 TaskManager의 작업을 다른 TaskManager로 재분배할 수 있음
  • SQL 문법만으로도 주요 스트리밍 처리 기능을 다룰 수 있음
    • SELECT로 데이터 형태를 변환하고 WHERE로 레코드를 필터링함
    • JOIN으로 여러 스트림을 결합하고 UNION으로 스트림을 합칠 수 있음
    • tumbling, hopping(sliding), session window 같은 윈도우 처리를 지원함
    • 이벤트 타임 처리와 워터마크로 지연 데이터 허용 범위를 설정할 수 있음
  • UDF와 커스텀 커넥터로 기본 기능 밖의 요구사항도 확장함
    • 기존 레거시의 대부분은 Redis SET 또는 INCR 커맨드를 사용하는 패턴이었고, Flink 공식 Redis Connector가 없어 Redis Connector를 직접 작성해 사용함
    • 당시 ARRAY 타입의 교집합을 구하는 빌트인 함수가 없어 UDF로 구현해 쿼리에 사용함

ksqlDB·Spark Structured Streaming과의 비교

  • ksqlDB는 사내에서 Kafka를 위해 사용하는 Confluent 플랫폼에 포함돼 있었고, 전사 사용 사례도 있었음
  • 하지만 상태 저장 스트리밍 처리의 HA 동작에서는 비효율이 있다고 판단함
    • 상태 저장 연산 failover 시 state 변화 기록인 changelog를 모두 replay해야 해 failover 시간이 오래 걸릴 수 있음
    • 처리 스트림 복제본을 두고 changelog를 내부 state에 계속 업데이트하는 방식은 복제본에서도 같은 연산을 수행하므로 리소스가 두 배로 소모될 수 있음
    • 관련 내용은 Configuring ksqlDB for High Availability | Confluent Developer에서 확인 가능함
  • Spark Structured Streaming은 Spark SQL 엔진 기반 스트리밍 처리 엔진임
    • 사내 사용 사례가 있고 UDF와 Custom Sink 작성이 가능함
    • Flink보다 규모가 크고 잘 구축된 에코시스템을 갖춤
  • Spark는 마이크로 배치 단위로 동작해 레코드 단위 지연이 발생할 수 있고, 실시간 처리가 중요한 상황에서는 Flink보다 불리할 수 있음
  • 팀 내 Spark 경험이 거의 없었고 Custom Sink 작성도 필요했기 때문에 Spark를 선뜻 선택하기 어려웠음

클러스터 환경 구축

  • 로컬에서는 Flink 공식 웹페이지에서 바이너리를 내려받고 {FLINK_HOME}/bin/start-cluster.sh로 클러스터를 실행할 수 있음
  • {FLINK_HOME}/bin/sql-client.sh를 실행하면 Flink SQL CLI가 열리고, SELECT 1; 같은 테스트 쿼리를 제출할 수 있음
  • 쿼리 제출 후 Flink 웹 UI에서 제출한 쿼리가 Job으로 변환되어 실행된 것을 확인할 수 있음
  • 2022년 말 Flink SQL Gateway가 릴리스되면서 HTTP 기반 쿼리 제출이 가능해짐

Kubernetes 기반 운영 아키텍처

  • 사내 대부분의 서비스가 Kubernetes 위에서 동작하므로 Flink SQL Cluster도 Kubernetes 위에 구성함
  • 기존 Flink App들은 모두 Application mode로 배포·운영되고 있었음
    • 각 애플리케이션마다 별도 클러스터를 띄우는 방식임
    • Kubernetes에서는 앱마다 JobManager Pod와 TaskManager Pod를 각각 띄워 동작함
    • 앱 간 독립성과 격리, 작업별 설정과 종속성 관리 측면에서 유리했음
  • Flink SQL은 이미 떠 있는 클러스터에 Job을 제출하므로 JobManager와 TaskManager를 Session mode로 띄워야 했음
  • 클러스터는 Stand Alone Cluster on Kubernetes 가이드를 기반으로 구성함
  • HA 환경은 High-Availability with Standalone Kubernetes 설정을 참고했고, high-availability.storageDir로는 s3를 사용함
  • Native Kubernetes 방식은 제공 shell script로 클러스터를 띄우기 때문에, deployment 설정을 직접 정의해 배포하는 사내 인프라 환경에는 적합하지 않다고 판단함

HA와 S3 연동 설정

  • HA 및 S3 연동을 위해 config.yaml에 다음 설정을 사용함
high-availability.type: kubernetes
high-availability.storageDir: s3://{s3-path-for-flinksql-recovery}
kubernetes.cluster-id: {cluster-id}
kubernetes.namespace: {k8s-namespace}


# namespace 내의 service account 를 통해 Kubernetes cluster 에 접근할 수 있도록 권한을 부여하는 작업이 필요할 수 있습니다.
kubernetes.service-account: {k8s-service-account-for-flinksql}
  • HA 환경에서는 JobManager pod를 두 개 띄우며, 서로의 주소가 달라야 리더 선출 로직 등이 정상 동작함
  • JobManager 컨테이너 실행 인자는 다음처럼 설정함
args: ["start-foreground", "-D", "jobmanager.rpc.address=$(POD_IP)"]
  • 이 설정을 통해 Kubernetes ConfigMap에 현재 리더로 선출된 JobManager pod 정보와 현재 실행 중인 Job ID 등이 저장되어 HA에 활용됨

GitOps 방식 쿼리 배포

  • Flink에서 직접 제공하는 Flink SQL용 웹 UI나 전용 툴은 아직 없음
  • Hue 연동 사례를 PoC로 검토했지만, 당시 Flink SQL Gateway 버전 호환성 이슈로 추가 개발이 필요했고 개발 환경 구성에도 시간이 많이 걸렸음
  • 사내에서 GitOps 패턴을 많이 사용하고 있어, 쿼리를 배포하거나 Job을 중단시키는 GitHub Actions를 구현함
  • Repository 안에 Job별 폴더를 만들고 실행할 쿼리를 SQL 파일로 모아둠
  • GitHub Actions는 폴더 이름을 받아 쿼리를 추출할 SQL 파일을 특정함
  • 구현은 Flink SQL Gateway REST API를 호출하는 방식이며, 간단하고 테스트하기 쉬운 Python으로 작성함

운영 사례와 장애 대응

  • JobManager가 실패한 경험은 없었지만, HA 설정상 JobManager가 실패하면 다른 JobManager가 리더로 선출되어 작업을 이어갈 수 있음
  • TaskManager는 가끔 실패했으며, 대부분 Kubernetes QoS 정책으로 Pod가 재시작되는 사례였음
  • 일부 TaskManager가 실패해도 작업이 다른 TaskManager로 재분배되어 계속되는 것을 확인함
  • 쿼리 실패는 대부분 비정상 데이터 인입이나 컴퓨팅 자원 부족 때문에 발생함
    • JSON 데이터를 읽을 때 잘못된 JSON 포맷은 json.ignore-parse-errors 옵션으로 에러 데이터를 무시할 수 있음
    • JSON_VALUE로 특정 path의 데이터를 추출할 때 값이 없거나 타입이 달라 발생하는 에러는 DEFAULT {VALUE} ON ERROR로 기본값을 설정할 수 있음
    • TaskManager CPU가 100%를 넘거나 메모리가 부족한 경우에는 TaskManager 리소스를 늘리거나 쿼리 parallelism을 늘린 뒤 재배포함
  • 클러스터 설정 변경이나 UDF 추가로 클러스터를 재시작할 때 일부 Job이 실패하는 경우가 있었음
    • 원인은 Job timeout이나 retry 설정이 적절하지 않은 경우가 많았음
    • Job이 너무 빨리 재시도를 종료하지 않고 클러스터 재시작 후 안정될 때까지 재시도하도록 timeout과 retry 설정을 수정함

쿼리 변경과 state 복원 제약

  • 쿼리 조건을 수정해 다시 배포할 때 savepoint로 state를 복원할 수 있는 경우는 조건식 값 변경 같은 아주 간단한 수정에 한정됨
  • window 조건이 바뀌면 state도 변경되어 호환성 유지가 어려워지고, savepoint를 이용한 복원이 어려울 수 있음
  • state를 유지해야 하는데 요구사항이 자주 바뀐다면 직접 앱을 작성하는 방식이 더 나을 수 있음

모니터링 포인트

  • Flink는 기본 제공 metric이 많아, 사내 모니터링 인프라와 적절한 Metric Reporter가 있으면 모니터링 환경을 쉽게 구성할 수 있음
  • numRunningJobs는 클러스터에서 현재 실행 중인 Job 개수를 나타내며, 값이 갑자기 감소한 채 유지되면 실패한 Job이 있다고 판단할 수 있음
  • taskmanager.cpu.load와 taskmanager.memory.used로 클러스터 리소스 사용량을 파악할 수 있음
  • busyTimeMsPerSecond로 TaskManager가 얼마나 바쁜지 Job별로 확인 가능함
  • Kafka를 source로 사용하는 경우 records-lag-max로 데이터 지연 상태를 빠르게 확인할 수 있음

예제: Kafka 로그인 이벤트 윈도우 집계

  • Appendix 예제는 Kafka에서 이벤트를 받아 10초마다 지난 1분간의 로그인 이벤트 수를 Kafka로 발행함
  • 입력 데이터는 JSON 포맷이며 event_time, event_type, data.user_id 필드를 포함함
  • 쿼리는 pipeline.name, parallelism.default, table.exec.state.ttl을 설정함
  • 입력 테이블 login_event는 Kafka connector와 JSON format을 사용하고, json.ignore-parse-errors를 true로 설정함
  • row_time은 event_time에서 생성하며, 워터마크는 현재까지 관측된 이벤트 타임 기준 최대 5초까지 늦게 도착하는 이벤트를 처리하도록 설정함
  • 출력 테이블 windowed_login_count는 Kafka topic으로 결과를 발행하고, proc_time AS PROCTIME() 필드를 포함함
  • HOP(row_time, INTERVAL '10' SECOND, INTERVAL '1' MINUTE)로 10초 간격의 1분 hopping window를 만들고, COUNT(*)로 window 내 레코드 수를 집계함

운영 결과와 남은 개선점

  • 팀은 기존 Flink 경험을 활용해 이전보다 쉽고 빠르게 여러 기능을 추가할 수 있었음
  • 생산성과 운영 효율성 측면에서 만족스러운 결과를 얻었다고 평가함
  • 도입 후 약 1년 동안 별다른 운영 작업 없이 안정적으로 동작함
  • 현재는 점차 확대 운영 중임
  • 쿼리 재배포와 클러스터 설정 변경에는 불편함이 남아 있으며, GitOps Controller 패턴 구현을 통한 쿼리 배포 환경 개선을 계획함

함께 보면 좋은 글 β

flgkselql98   [-]

flink와 같은 분산 시스템은 rack을 2~3개 유지함으로써 HA를 유지해야하는데 kubernetes를 연동함으로써 HA를 보장한 것 같네요. 근데 결국 kube slave node에 대한 리소스도 고민해야할텐데 flink만 띄우는 node를 구성한건지 생각이 되네요 (flink 부하시에 slave node 다운 이슈가 있을 듯)
그런 관점에서 kubernetes를 쓰는 장점이 있을까요?

또한 flink에서 윈도우 함수를 쓰게 되면 그 사이 데이터는 메모리에 유지됨으로써 sql join문이 동작하는 건데 trade - off 관점에서 보았을때 flink는 좋은 선택지일까 생각이 듭니다. 시간이 지날수록 거대해지는 sql + job이 죽게되면 발생하는 엄청난 일..

저도 최상단 data source에서 join이 필요한 상황일때 flink를 안쓰고 어떤 방식으로 application level로 내려서 처리 할 수 있을지 고민이 되네요.

답변달기