스트림 데이터 처리를 진행하기 전에 반드시 이해해야 하는 중요한 개념이 있다.

끊임없이 유입되는 데이터 스트림은 본질적으로 경계가 없다. 이러한 특성으로 인해 스트림 전체에 대한 통계 작업을 수행하는 것은 불가능하다. 예를 들어, 특정 시간을 기준으로 얼마나 많은 데이터가 통과했는지, 데이터의 최대값·최소값·평균값 또는 누적값 등을 계산할 수 없는 것이다.

그러나 스트림에서 일정 구간을を切り取込んで 그 범위 안에서의 통계를 구하는 것은 충분히 가능하다. 이러한 구간切り取り 방식은 크게 두 가지 유형으로 구분할 수 있다.

이제 실제 코드를 통해 윈도우 연산의 구체적인 구현 방법을 살펴보도록 하자.

1. Tumbling Time Window (비중복 시간 윈도우)

실생활 예시

교차로에서 통과하는 차량 수를 계산한다고 가정해보자. 끊임없이 흐르는 차량 흐름 전체를 한 번에 계산하는 것은 불가능하다. 그러나 15초 단위로 구분하면 첫 번째 15초에는 2대, 두 번째 15초에는 3대, 세 번째 15초에는 1대와 같이 특정 시점의 차량数を集計할 수 있다.

구현 코드

// 데이터 수집을 위한 소켓 서버 실행
// 터미널에서 다음 명령어 입력
nc -lk 9999

// 테스트용 데이터 전송 (센서ID, 차량수)
// 각 줄마다 Enter 키 입력
9,3
9,2
9,7
4,9
2,6
1,5
2,3
5,7
5,4

import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.api.windowing.time.Time

object TrafficWindowExample {
  def main(args: Array[String]): Unit = {
    // 1. 스트림 실행 환경 초기화
    val streamEnv = StreamExecutionEnvironment.getExecutionEnvironment

    // 2. 소켓을 통한 입력 데이터 소스 정의
    val sourceData = streamEnv.socketTextStream("localhost", 9999)

    // 3. 데이터 파싱 및 모델 클래스 매핑
    case class TrafficRecord(intersectionId: Int, vehicleCount: Int)
    val parsedData: DataStream[TrafficRecord] = sourceData.map {
      line => {
        val fields = line.split(",")
        TrafficRecord(fields(0).trim.toInt, fields(1).trim.toInt)
      }
    }

    // 4. 윈도우 집계 연산 수행
    // 각 교차로별로 5초 단위의 튜블링 윈도우 생성
    // 5초 동안 수집된 데이터에 대해 차량 수 합산
    val aggregatedResult: DataStream[TrafficRecord] = parsedData
      .keyBy("intersectionId")
      .timeWindow(Time.seconds(5))
      .sum("vehicleCount")

    // 5. 결과 출력
    aggregatedResult.print()

    // 6. 스트림 처리 작업 실행
    streamEnv.execute(this.getClass.getSimpleName)
  }
}

2. Sliding Time Window (중복 시간 윈도우)

슬라이딩 윈도우는 윈도우 크기보다 작은 간격으로 윈도우를 이동시킨다. 이를 통해 최근 N초 동안의 데이터를 중복 포함시키면서 통계를 산출할 수 있다.

import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.api.windowing.time.Time

object SlidingWindowExample {
  def main(args: Array[String]): Unit = {
    // 1. 스트림 실행 환경 초기화
    val streamEnv = StreamExecutionEnvironment.getExecutionEnvironment

    // 2. 소켓을 통한 입력 데이터 소스 정의
    val sourceData = streamEnv.socketTextStream("localhost", 9999)

    // 3. 데이터 파싱 및 모델 클래스 매핑
    case class TrafficRecord(intersectionId: Int, vehicleCount: Int)
    val parsedData: DataStream[TrafficRecord] = sourceData.map {
      line => {
        val fields = line.split(",")
        TrafficRecord(fields(0).trim.toInt, fields(1).trim.toInt)
      }
    }

    // 4. 슬라이딩 윈도우 연산 수행
    // 윈도우 크기: 10초, 슬라이딩 간격: 5초
    // 매 5초마다過去 10초 데이터에 대한 집계 수행
    val aggregatedResult: DataStream[TrafficRecord] = parsedData
      .keyBy("intersectionId")
      .timeWindow(Time.seconds(10), Time.seconds(5))
      .sum("vehicleCount")

    // 5. 결과 출력
    aggregatedResult.print()

    // 6. 스트림 처리 작업 실행
    streamEnv.execute(this.getClass.getSimpleName)
  }
}

3. Tumbling Count Window (비중복 개수 윈도우)

시간 기반이 아닌 데이터 개수 기반의 윈도우도 활용할 수 있다. 특정 키를 가진 레코드가 N개 수집될 때마다 윈도우를 트리거한다.

// 데이터 수집을 위한 소켓 서버 실행
nc -lk 9999

// 테스트용 데이터 전송
9,3
9,2
9,7
4,9
2,6
1,5
2,3
5,7
5,4

import org.apache.flink.streaming.api.scala._

object CountWindowExample {
  def main(args: Array[String]): Unit = {
    // 1. 스트림 실행 환경 초기화
    val streamEnv = StreamExecutionEnvironment.getExecutionEnvironment

    // 2. 소켓을 통한 입력 데이터 소스 정의
    val sourceData = streamEnv.socketTextStream("localhost", 9999)

    // 3. 데이터 파싱 및 모델 클래스 매핑
    case class TrafficRecord(intersectionId: Int, vehicleCount: Int)
    val parsedData: DataStream[TrafficRecord] = sourceData.map {
      record => {
        val fields = record.split(",")
        TrafficRecord(fields(0).trim.toInt, fields(1).trim.toInt)
      }
    }

    // 4. 카운트 기반 튜블링 윈도우 연산
    // 각 교차로별로 5개의 메시지가 도착할 때마다 집계 수행
    val aggregatedResult: DataStream[TrafficRecord] = parsedData
      .keyBy("intersectionId")
      .countWindow(5)
      .sum("vehicleCount")

    // 5. 결과 출력
    aggregatedResult.print()

    // 6. 스트림 처리 작업 실행
    streamEnv.execute(this.getClass.getSimpleName)
  }
}

4. Sliding Count Window (중복 개수 윈도우)

슬라이딩 개수 윈도우는 윈도우 크기와 슬라이딩 간격을 다르게 설정하여, 최근 N개 데이터 중 최근 M개 데이터를 기준으로 연산을 수행한다.

import org.apache.flink.streaming.api.scala._

object SlidingCountWindowExample {
  def main(args: Array[String]): Unit = {
    // 1. 스트림 실행 환경 초기화
    val streamEnv = StreamExecutionEnvironment.getExecutionEnvironment

    // 2. 소켓을 통한 입력 데이터 소스 정의
    val sourceData = streamEnv.socketTextStream("localhost", 9999)

    // 3. 데이터 파싱 및 모델 클래스 매핑
    case class TrafficRecord(intersectionId: Int, vehicleCount: Int)
    val parsedData: DataStream[TrafficRecord] = sourceData.map {
      record => {
        val fields = record.split(",")
        TrafficRecord(fields(0).trim.toInt, fields(1).trim.toInt)
      }
    }

    // 4. 카운트 기반 슬라이딩 윈도우 연산
    // 윈도우 크기: 5개, 슬라이딩 간격: 3개
    // 데이터 3개마다 최근 5개 데이터를 기준으로 집계 수행
    val aggregatedResult: DataStream[TrafficRecord] = parsedData
      .keyBy("intersectionId")
      .countWindow(5, 3)
      .sum("vehicleCount")

    // 5. 결과 출력
    aggregatedResult.print()

    // 6. 스트림 처리 작업 실행
    streamEnv.execute(this.getClass.getSimpleName)
  }
}

윈도우 연산 정리

Flink에서 윈도우 처리 방식은 다음과 같이 분류된다:

1. 윈도우 분류 기준 (분할 방식)

• Time Window:时间来区分窗口
• Count Window:数据个数来区分窗口

2. 윈도우 속성

• Size (크기): 윈도우 전체 범위
• Interval (간격): 윈도우 이동 간격

3. 윈도우 유형 결정 조건

• Size = Interval: 튜블링 윈도우 (비중복)
• Size > Interval: 슬라이딩 윈도우 (중복 데이터 포함)
• Size < Interval: 데이터 유실 발생 (권장하지 않음)

4. 네 가지 기본 윈도우 조합