[GCP] Apache Beam 알아보기

Apache Beam과 Dataflow

13 min readAug 16, 2021

Apache Beam

Apache Beam은 구글에서 개발하여 2016년에 오픈소스로 공개한, ETL, batch, streaming 파이프라인을 처리하기 위한 unified programming model이다. Beam의 가장 큰 특징은 다양한 랭귀지와 다양한 runner를 지원한다는 것이다. Beam SDK를 통해 다양한 runner들(분산처리 백앤드: Apache Flink, Apache Samza, Apache Spark, Google Cloud Dataflow)에서 데이터를 처리할 수 있게 해준다.

바로 이 특징이 Beam의 탄생의 비젼이다. 빅데이터 처리 생태계는 Mapreduce, Spark, Flink 등 빠르게 세대가 바뀌고 매년 더 발전된 분산처리엔진이 나오고 있는데, 이 트랜드를 따라가면서 각자의 장단점에 따라 하나를 선택하기도 쉬운 일이 아닐 뿐더러, 오늘의 최신 기술이 내일이면 구세대 기술이 된다.. 이제 Beam을 통해 원하는 랭귀지로 개발하고, 러너는 원하는 것으로 선택할 수 있도록 하자는 것이다.

원하는 언어로 작성해서 원하는 러너에서 수행하도록 해주자는 원대한 꿈(from Overview of Apache Beam)

SDKs

Java, Python, Go 세가지 언어의 SDK를 제공하므로 선호하는 언어를 사용해 데이터 프로세싱 파이프라인을 개발할 수 있다.

Beam Pipeline Runners

빔에서 지원하는 러너들은 꽤 다양하다.

Direct Runner
Apache Flink
Apache Samza
Apache Spark
Google Cloud Dataflow

위 목록 외에도 몇가지 러너를 더 지원한다. 한번 개발한 파이프라인을 다양한 분산처리엔진에서 실행할 수 있다는 점에서 뛰어난 이식성이 매력적이다. 다만 러너에 따라 지원하는 기능이 일부 다를 수 있으므로 개발 및 처리엔진 선택 전에 이 부분을 고려해야한다. 러너별 지원항목은 링크의 capability matrix를 참고하자.

위의 런타임 중에서 Beam의 가장 대표적인 런타임은 Google Cloud의 Dataflow다. Dataflow에 대해서 살짝 살펴보자.

Dataflow

Dataflow는 Apache Beam SDK를 활용해 배치와 스트리밍 데이터 프로세싱 파이프라인을 구현할 수 있도록 해주는 GCP의 서비스이다. 매니지드 서비스이므로, 서버와 인프라에 대한 고려 없이 서버리스로 데이터 파이프라인을 개발할 수 있다는 장점이 있다.

‘데이터 파이프라인 개발’이라는 의미에서 Airflow나 Oozie와 같은 Workflow Orchestration tool 을 생각할 수도 있는데, 이보다는 Apache Spark나 Flink 등 데이터 프로세싱 엔진의 일종이다. (참고로 GCP 내 워크플로우 개발을 위해서는 Apache Airflow를 매니지드 서비스로 구현한 Cloud Composer라는 서비스가 있다.)
Dataflow는 Apache Beam이라는 오픈소스 데이터 프로세싱 파이프라인 개발 툴킷의 런타임(Runtime) 중 하나이므로, Dataflow 활용을 위해서는 Apache Beam 의 API와 코드에 대해서 알아야 한다.

Objects of Beam

Data and Transformation

Beam의 데이터 오브젝트에 대해서 알아보자.
먼저 Beam의 데이터파이프라인을 구성하는 두가지 주요 오브젝트는 Pcollections와 PTransfroms다.

PCollections
Beam의 파이프라인 안의 모든 데이터는 PCollection안에서 산다. (Pcollection은 Parallel Collection을 줄인 용어다.)
Pcollection은 분산데이터셋이며, immutable한 속성을 가진다.
PTransforms
데이터를 트랜스포메이션 하는 작업은 PTransforms(Parallel Transform)라는 function을 통해 한다. input으로 PCollections을 받고 변형하여 PCollections output을 리턴한다.

Pardo

Pardo는 Beam의 가장 일반적인 parallel processing transform이다.
Pardo는 분산처리 트랜스포메이션의 Map — Shuffle — Reduce 단계중 Map에 가깝다. 아래와 같은 일반적인 데이터프로세싱 연산에 모두 사용한다.

Filtering a data set
Formatting or type-converting each element in a data set
Extracting parts of each element in a data set
Performing computations on each element in a data set

아래는 Pardo를 실행하기 위한 샘플 스니펫이다.
(Beam에서 파이프라인 내 스텝의 연결을 위한 delimiter는 파이프(|)를 쓴다.)

이번에는 Map 연산의 예시를 하나 보자.

파이썬에서는 위 연산을 위해서 일반적으로 map이나 list comprehension을 통해서 구현할 것이다.

위 연산을 Beam의 Map transform을 통해 구현할 수 있다.

Beam은 functional programming 패러다임에 따라 디자인 되었다. 따라서 loop대신, PCollection내의 각 데이터의 처리를 위해 PTransform 함수를 연결해서 사용할 수 있다.
(파이프라인 내에서 ‘abc’ >> 와 같은 형태로 comment를 사용할 수 있다.)

위의 코드 및 더 많은 transform 관련 예시는 Apache Beam의 공식 문서에서 제공하는 Colab에서 살펴볼 수 있다.

Windowing

Windowing은 PCollection을 타임스탬프나 어떤 기준 요소로 잘라서 보는 것을 말한다. 어떤 PCollection을 논리적 윈도우로 잘라볼 수 있다. 각 요소들은 하나 혹은 그 이상의 윈도우에 속해서 GroupByKey 또는 Combine등의 Aggregation연산 등에 활용될 수 있다.
Windowing 방식에는 세가지가 있다:
1. Fixed time windows
2. Sliding time windows
3. Session windows
순서대로 한번 살펴보자.

1. Fixed time windows

가장 기본적인 형태의 윈도우. 지정된 타임스탬프 단위의 고정된 시간으로 윈도우를 잘라내는 형태로 활용한다.

30초 단위의 fixed window를 설정하는 코드 스니펫.

from apache_beam import window
fixed_windowed_items = (
    items | 'window' >> beam.WindowInto(window.FixedWindows(60)))

2.Sliding time windows

위와 같이 timestamp 등 시간 기준으로 데이터 스트림을 잘라내는데, 윈도우간 겹치게 오버랩을 할 수 있다.
윈도우의 길이는 duration이라고 하며, 각 윈도우가 시작되는 period라고 한다. 아래 예시에서는 duration이 60초짜리 윈도우를 30초마다 시작되게 하였으므로 period는 30초이다.

3. Session windows

마지막으로 세션 윈도우는 윈도우 사이에 갭이 생길 수 있는 데이터의 형태에 활용할 수 있다. 정해진 minimum gap duration보다 작은 간격을 두고 들어온 데이터는 동일한 윈도우로 처리한다. 클릭스트림 로그, 유저 로그 등에서 유저별로 세션을 만들어서 기록할 때 활용할 수 있다.

Watermark and Late data

워터마크는 스트리밍 처리 시 늦게 도착하는 데이터의 이슈를 풀기 위한 재미있는 컨셉이니 한번 보자.
특정 타임스탬프 안에 들어오는 데이터를 윈도우에 넣는 형태로 데이터를 프로세싱할 때, 네트워크, 클라이언트 등의 이슈로 데이터가 지연 도착하는 경우가 발생할 수 있다. 지연이 심해지면, 원래 해당 데이터가 속해야 하는 윈도우의 데이터 프로세싱이 끝난 이후 도착하는 경우도 있을 수 있다.(12:30:00에 윈도우가 종료되었는데 12:29:59 타임스탬프 데이터가 12:31:00 도착한 경우)
이런 데이터의 처리를 위해서 Beam에서는 Watermark라는 기능을 제공한다.
윈도우가 종료된이후라도 계속 데이터가 들어오면 Watermark를 뒤로 미루는 식으로 다이나믹하게 Watermark를 세팅할 수 있다. Watermark내에 들어오는 데이터는 지연도착으로 따로 분류되지 않고 윈도우 내에 속할 수 있게 된다.

Watermark: 워터마크 타임스탬프가 지정된 시간을 지나면, 시스템은 이보다 늦게 들어오는 데이터는 없을 것이라 인식한다. 즉 마지막 지각생까지 처리가 되었다고 가정한다. 그 이후에 들어오는 데이터는 설정에 따라 버려지거나 지연 처리 하게 된다.
Late data: 워터마크 이후에 들어오는 데이터를 late data라고 한다.

Beam의 디폴트 설정은 윈도우의 끝 지점과 워터마크 지점이 동일하기 때문에, late data를 허용하지 않는다.
아래 late data를 허용하기 위한 샘플 스니펫을 보자.

pc = [Initial PCollection]pc | beam.WindowInto(
    FixedWindows(60),
    trigger=trigger_fn,
    accumulation_mode=accumulation_mode,
    timestamp_combiner=timestamp_combiner,
    allowed_lateness=Duration(seconds=2*24*60*60)) # 2 days

Trigger

스트리밍 처리된 데이터를 윈도우로 그룹핑 하기 위한 기준을 트리거 라고 한다. PCollection의 트리거 기준은 자유롭게 세팅할 수 있다.

기본적인 몇가지 pre-built 트리거만 살펴보면:

Event time triggers
Beam의 디폴트 트리거. 데이터의 타임스탬프 기준으로 윈도우 그룹핑
Processing time triggers
파이프라인 내의 프로세싱 타임 기준 트리거.
Data-driven triggers
윈도우내 데이터를 체크 해서 트리거. 현재는 데이터 개수 기준 트리거링만 제공함.
Composite triggers
몇가지 트리거를 조합해서 생성하는 트리거.

Beam의 장단점

Beam이 Spark나 Flink 같은 시스템 대비 더 나은 점은 뭘까?

마침 궁금한 질문을 누군가 Stack Overflow에 물어보았고, Beam의 PMC인 분이 답변을 달아주었다.(link: Stack Overflow)

Unifying batch and streaming:
많은 시스템이 배치와 스트리밍을 별도의 API로 처리한다. Beam은 배치와 스트리밍 사이에 별도의 러닝커브나 코드 rewriting이 필요없다.
배치 수행하다가 그냥 내일부터 low latency 스트리밍으로 처리하고 싶으면 같은 코드로 재설정만 하면 된다. (Flink나 Spark Structured Streaming도 가능한 부분이다.)
APIs that raise the level of abstraction:
API의 추상화 레벨을 올렸다. 예를 들어, Beam의 ParDo액션을 통해 데이터를 처리하면, 나머지 각 러너에 따른 기본적인 성능 최적화는 Beam이 알아서 해준다.
Portability across runtimes:
간편한 런타임 이식성. 데이터 파이프라인과 런타임이 분리되어 있으므로, 온프렘에서 클라우드로 옮긴다던가, 레거시 시스템에서 cutting-edge 시스템 이식이 간편하다. 런타임만 바뀌면 되니까. 같은 이유로 한 시스템으로 온프렘과 클라우드 등에 여러 러너를 가진 하이브리드 시스템으로 쓰기에도 유리하다. (솔깃?)

위와 같은 장점을 갖기 위해 수반되는 단점도 있어 보인다.

추상화 레벨을 한 레이어 올리고 그 아래 처리엔진들이 수행되는 만큼, 당연히 스파크 등 레이어 아래에 있는 프로젝트의 모든 기능을 지원하기 힘들며 이에 따른 커스터마이징된 옵티마이제이션, 튜닝에는 제약이 존재할 수 밖에 없을 것이다.

Spark 등의 메이저 아파치 프로젝트 대비해서는 상대적으로 작은 오픈소스 커뮤니티를 갖고 있다. 기능 개발 속도나 트러블슈팅 지원, Stack Overflow등의 정보량에 있어서도 상대적으로 제약이 있을 수 있다.