이번 포스팅에서는

이미지와 컨테이너 관리하기

1
2
3

docker ps

docker run은 이미지를 기반으로 새로운 컨테이너를 띄운다.
따라서 별도의 수정이 없는 경우에는 docker start 명령으로 기존에 중지된 컨테이너를 재 사용할 수 있다.

docker run으로 컨테이너를 시작하면 container가 background가 아닌 foreground에서 실행된다. 반면 docker start는 background에서 container가 실행된다.

Attached mode(docker run default) , Detached mode(docker start)

Detached mode로 실행된 컨테이너의 경우에는 애플리케이션에서 발생한 로그 기록들을 실시간으로 확인(console 확인)할 수 없으며, Attached mode로 실행된 컨테이너의 경우에는 애플리케이션의 실행과정에서 발생한 로그 기록들을 실시간으로 console을 통해 확인할 수 있다.

컨테이너의 출력결과를 수신한다.

이번 포스팅에서는 Docker의 이미지 & 컨테이너(코어 빌딩 블록)에 대해 학습한 내용을 정리한다.

Images and Container

• Pre-Built & Custom Images

• Create & Manage containers

Images와 Container의 관계

• Container : The running “unit of software”

• Images : Templates / Blueprints for containers

Contains code + required tools/runtimes

이미지를 사용하여 여러 컨테이너(running instance)를 생성할 수 있다.

Finding / Creating Images

• 이미지는 기존에 생성된 이미지를 사용하거나, Dockerhub에 업로드 되어있는 pre-built 이미지를 사용할 수 있다.

1

docker run --name node-image node

위의 명령을 통해 node pre-built 이미지를 Docker hub를 pull해서 로컬에 다운받을 수 있다.

1

docker ps -a

위의 명령을 통해서 도커가 생성한 모든 컨테이너 정보를 확인할 수 있다.

1
2

docker run -it node
# 쉘을 통해서 node container에 접근

나만의 CUSTOM DOCKER IMAGE를 생성하기 위해 Dockerfile을 작성할 수 있다.

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29

# 운영체제 레이어 SETTING
FROM node

WORKDIR /app

# 첫번째 .(dot) : 컨테이너 외부 경로로, Dockerfile이 위치한 경로를 의미한다.
# 두번째 .(dot) : 자체 내부 시스템
# 만약에 /app 폴더가 존재하지 않는경우에는 이미지와 컨테이너에 생성된다.
COPY . /app

# 기본적으로 도커 컨테이너 및 이미지의 WORKDIR에서 아래 명령이 실행된다.
RUN npm install

# DOCKER 컨테이너는 외부 로컬 시스템과 격리되어있다.
# 자체 내부 네트워크도 있기 때문에 외부 로컬 시스템에 노출시켜줘야 한다.
EXPOSE 80

# 서버 시작
# 아래 명령도 틀린건 아니지만, 이 이미지가 빌드될 때마다 실행되기 때문에 
# 아래와 같이 실행하는 것은 올바르지 않다.
# 위에서 작성한 모든 것은 이미지 설정을 위한 명령으로, 도커 명령어이다.
# 이미지는 컨테이너의 템플릿이어야 한다. (이미지를 실행하는 것이 아닌, 이미지를 기반으로 컨테이너를 실행하는 것)
# RUN node server.js 

# 컨테이너를 실행하는 경우에만 서버를 실행하기를 원하기 때문에
# 아래와 같이 명령을 실행하는 것이 맞다.
# RUN과 CMD의 차이는 CMD는 이미지가 생성될 때 실행되지 않고, 컨테이너가 실행될때 실행된다.
# 만약 명시된 CMD 명령이 없다면, BASE IMAGE의 CMD 명령이 실행되며, BASE IMAGE가 없고, CMD 이미지가 없다면 에러 메시지가 팝업된다.
CMD ["node", "server.js"]

그 다음 STEP은 위에서 작성한 Dockerfile의 스크립트 내용을 기준으로 Docker image를 만들어야 한다.

1

docker build --tag node-custom-app .

아래 명령으로 생성한 이미지(node-custom-app)를 사용하여 node-custom-app-container 도커 컨테이너를 80번 포트로 portforwarding하여 백그라운드로 실행한다.

1
2

# 실행할 Docker 이미지의 full name 말고, 만약 이름이 유일하다면, 부분이름으로 지정해서 사용할 수 있다.
docker run --name node-custom-app-container -p 80:80 -d node-custom-app

Dockerfile에서 EXPOSE 80은 문서화하는 목적으로 작성되었기 때문에 실제 컨테이너를 실행할때 위와같이 -p option을 통해 portforwarding 해줘야 한다. (EXPOSE 명령은 선택사항이지만 모범적인 사용법)

여지까지 기존 이미지인 노드 이미지를 기반으로 하는 커스텀 이미지를 docker build 명령을 통해서 생성하였고, docker run 명령을 통해서 해당 이미지를 기반으로 컨테이너를 띄웠다.

CUSTOM APP 코드 수정

커스텀 앱의 코드 일부를 수정했다면, 생성한 커스텀 이미지를 다시 빌드하여, 코드 복사과정을 다시 실행해야 한다.

한 번 빌드된 이미지는 잠겨있으며, 읽기 전용이 된다.

이미지는 레이어 기반 아키텍처(모든 명령이 레이어로 구성)로 되어있기 때문에 이미지를 빌드하는 과정에서 과거 실행되었던 부분에 대해서는 모든 명령 결과를 캐시하고, 이미지를 다시 빌드할때 명령을 다시 실행할 필요가 없으면, 캐시된 결과를 사용한다.

만약 특정 레이어가 변경되었다면, 변경된 레이어 후의 모든 레이어가 다시 빌드된다. 따라서 Dockerfile 내 레이어를 최적화 시키기 위해서는 우선적으로 package.json파일 개별적으로 복사하고, npm install 명령을 실행하고 반복적으로 코드 수정이 발생하는 부분에 대해서는 별도로 복사를 하는 과정을 작성하도록 하는 것이 좋다.

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16

# 운영체제 레이어 SETTING
FROM node

WORKDIR /app

# 선 dependencies 파일 복사 및 설치
COPY package.json /app

RUN npm install


COPY . /app

EXPOSE 80

CMD ["node", "server.js"]

이 이미지를 기반으로 컨테이너를 실행하면, 컨테이너는 기본적으로 Dockerfile에 지정한 명령을 실행한 결과로 코드를 실행중인 애플리케이션인 이미지 위에 컨테이너 레이어를 새로 추가한다.

CUSTOM APP 코드 수정

이번 포스팅에서는 Docker에 대해 학습한 내용을 기록하고자 한다.

Docker

Docker는 컨테이너 기술로, 컨테이너를 생성하고 관리하는데 사용된다.

VM

VM과 비교했을때, VM 또한 도커와 같이 환경을 분리하고, 각 환경별로 구체적인 환경설정을 분리하는 것 또한 가능하다. 그리고 구성한 환경 설정을 공유할 수 있고, VM 간의 의존성을 낮출 수 있다.
위의 장점이 있지만, 환경 구성을 위해 각 각의 VM의 기반으로 OS를 설치해야 되기 때문에 시스템의 리소스가 낭비될 수 있으며, 퍼포먼스가 느려지고, 부팅 시간이 길어지는 단점이 있다.

Docker 구성

Docker 엔진 위에 복수의 컨테이너들을 띄워서 구성할 수 있다. 물론 Docker container 내에도 OS를 설치하지만 VM의 OS보다 훨씬 가볍고 빠르다. Share, Re-building, Distribution에 용이하며, apps와 환경을 Encapsulation한다.

Docker image

Container는 이미지를 기반으로 하기 때문에 이미지 생성을 우선 해야한다.
각 각의 이미지에는 개별 파일 시스템이 있기 때문에 Dockerfile이 위치한 경로에 있는 파일들을 Docker의 파일 시스템에 복사해서 처리해줘야 한다.

우선 본 포스팅에서 다루고자 하는 에러가 어떤 상황에서 발생하게 되었는지에 대해서 간략하게 설명을 하고, 어떤 접근으로 해결하려고 했는지에 대해서 기술해보려고 한다.

프로젝트 진행

회사에서 진행중인 프로젝트에서 백엔드를 구성하면서 간단한 CRUD 처리의 경우에는 JPA를 사용하고, 복잡한 쿼리 사용의 경우에는 MyBatis를 사용하기로 했다. 여기서 말하는 복잡한 쿼리란 간단한 CRUD 이외의 복잡한 쿼리를 말한다.
그냥 이렇게 하라고 하니깐 하는 게 아니라 이렇게 구성을 하면 어떤 이점이 있는지에 대해서 간단하게 짚고 넘어가도록 하자.

• Hibernate vs MyBatis

  과거 EJB2로 개발을 하던 당시 Gavin king이라는 사람이 사용자 친화적이지 않은 자바 애플리케이션 개발 방식을 좀 더 사용자 친화적이게 만들고자 개발을 하게 된 것이 바로 그 유명한 Hibernate이다. 이 Hibernate가 점차 인기가 많아지자 자바 진영에서 Gavin king을 영입해서 자바 ORM(Object Relational Mapping) 기술에 대한 표준 명세를 개발하도록 하였는데, 그것이 바로 JPA(Java Persistent API)이다.

  JPA는 ORM을 사용하기 위한 인터페이스를 모아둔 것으로, 이를 구현한 프레임워크 중 대표적으로 Hibernate가 있고, Spring에서는 대부분 Hibernate를 사용하고 있다.
  Spring에서는 JPA를 사용할 때 구현체들을 직접 다루지 않고, 구현체들을 좀 더 쉽게 사용하고자 추상화 시킨 Spring Data JPA라는 모듈을 시용하여 JPA 기술을 다룬다.

  그렇다면, JPA 하나만 사용해서 개발하면 만능일까? 찾아보니 대부분 단순한 CRUD의 처리를 하는 경우에는 JPA만을 사용해도 괜찮다고 한다. 하지만, 복잡한 통계나 정산관련 조회 쿼리가 포함된 경우, MyBatis로 처리하면 좀 더 개발자에게 편하다. 물론 JPA로도 복잡한 집계성 쿼리를 처리하는 것도 가능은 하지만 구현이 쉽지 않기 때문에 MyBatis를 사용하는 것이 낫다고 한다.

  따라서 JPA 또는 MyBatis만 사용하는 것이 아닌, 두 개를 적절히 조합해서 프로젝트를 진행하면 업무적으로 효율이 높아질 수 있다는 결론이 나온다.

  
  

• MyBatis의 사용

  MyBatis는 Mapper를 별도로 구성하며, SqlSession을 직접 사용하는 형태가 아닌 Mapper를 통해 처리한다. 기본적으로 Mapper를 사용하게 되면, Mapping 파일이 자동으로 Mapper의 단위가 되기 때문에 유지보수 및 관리에 용이하다.

이번 포스팅에서는 최근에 새롭게 접하게 된 ETL 툴인 Apache NiFi에 대해서 정리해보려고 한다.

NiFi의 기술적 배경

NiFi는 NAS라는 미국의 국가 안보국에서 Apache에 기부한 Dataflow 엔진으로, ETL툴의 일종이다.
NiFi는 과거에 NSA에 의해 개발되었다가 2014년 기술 전송 프로그램의 일부로, 오픈소스화된 나이아가라파일즈(NiagaraFiles)에 기반을 두고 있다.

NiFi는 분산환경에서 대량의 데이터를 수집, 처리한다. => FBP개념을 구현한 오픈소스 프로젝트
여기서 FBP란 사전에 Data Flow를 정의하고, 이를 지속적으로 유지하면서 데이터를 교환하는 프로그래밍 패러다임이다.
(마치 Apache Airflow에서 DAG를 작성해서 전체적인 Task flow를 구성한 뒤에 반복적인 데이터 처리를 자동화 시키는 것과 같은 맥락이다)

NiFi의 필요성

그래서 왜 NiFi가 필요할까? 앞서 이미 정의했듯이 NiFi는 ETL툴의 일종으로, 클러스터로 구성이 되어있기 때문에 대량의 데이터를 분산시켜서 처리할 수 있다.
NiFi는 A 시스템에서 B 시스템으로 데이터를 이관하는 것을 손쉽게 할 수 있도록 도와주는 서비스 툴로, 데이터를 이관하는 중간에 데이터를 변형(정제)할 수 있다. 그 외에도 관리 및 모니터링이 가능하다.

NiFi의 구성

NiFi는 FlowFile, Processor, Connection, 이 세 가지로 구성이 되어있다.

• [FlowFile]

  FlowFile은 NiFi가 인식하는 데이터 단위로, 속성과 내용이 Key/Value 혀태로 구성이 되어있다. Processor마다 이동시 복사본이 생성되어, 추적이 용이하다.

  ref. query를 작성할때 [table명] 대신에 FLOWFILE을 작성하게 되면, 파이프라인 상에서 흘러가는 파일 정보를 읽을 수 있다.

• [Processor]

  FlowFile을 수집, 변형 및 저장하는 기능이 있으며, 자주 사용되는 프로세서로는 http, kafka, db, ftp와 관련된 프로세서, 속성을 변경하는 updateattribute, 데이터를 합치는 mergecontent, 데이터를 분할하는 split, 데이터 타입을 변경하는 convert등이 있다.

• [Connection]

  각 Processor별로 연결해서 FLOWFILE을 전달하는 역할을 담당하고, FlowFile의 우선순위, 만료, 부하조절 기능도 제한하고 있다.

NiFi 아키텍쳐

• [Web Server]

  발생하는 이벤트를 모니터링, 소프트웨어를 시각적으로 제어하기 위해서 사용된다. (HTTP기반 구성요소)

• [Flow controller]

  NiFi 동작의 뇌 역할을 담당한다. NiFi 확장기능의 실행을 통제하고, 이를 위한 자원 할당을 스케줄링한다.

• [Extensions]

  NiFi가 다양한 종류의 시스템과 통신할 수 있게 하는 다양한 플러그인이다.

• [FlowFile Repository]

  NiFi가 현재 실행중인 FlowFile의 상태를 추적하고, 정비하기 위해 사용된다.

• [Content Repository]

  전송 대상의 데이터가 관리된다.

Configuration 살펴보기

더 많은 Airflow Worker가 필요하다면, 추가 machine에서 celery worker를 명령한다.

Flower

Airflow workers를 dashboard를 통해서 모니터링하기 위한 툴이다. Celery Executor를 사용하면, Flower에 접속해서 Celery Executor의 Administrator와 Airflow worker를 대시보드를 통해 모니터링 할 수 있다.

localhost:5555/dashboard

아래의 Worker dashboard를 살펴보면, Max concurrency 값이 16인 것으로 보아, 최대 16개의 Task를 동시에 실행하는 것이 가능하다. 이는 사용하는 PC의 리소스에 따라 줄일 수도 늘릴 수도 있다.

다음으로 Queues 메뉴는 유용하게 사용될 수 있는데, 특정 Task를 특정 worker로 라우팅되도록 할 수도 있다.
예를들어 높은 리소스를 소비하는 Task가 존재하고, 현재 하나의 높은 리소스 Worker를 가진다고 가정하면, queue를 생성해서 queue를 Worker에 붙이고, 높은 리소스 소비 Task를 생성한 queue로 보내서 해당 Worker만 해당 작업을 실행할 수 있도록 queue를 지정해서 사용할 수 있다.

Backfilling

DAG를 처음 실행하게 되면, scheduler는 자동으로 non-triggered DagRuns을 시작 날짜(start_date)와 현재(now) 사이 시점에서 실행하게 된다.
catch up mechanism은 자동으로 non-triggered DagRun을 마지막으로 실행된 날짜와 현재 시간 사이에서 실행할 수 있도록 허용한다.

예를들어, 만약에 DAG를 2일동안 중지시키고나서 DAG를 다시 시작했다면, 이 기간 동안 트리거되지 않은 DAG 실행에 해당하는 일부 DAG 실행이 발생합니다.
Backfilling mechanism은 historical DagRuns를 실행하도록 하는데, 예를들어 start date 이전의 기간에 DagRun을 실행할 수 있다.
방법은 Airflow DAG Backfill 명령을 실행하는 명령을 사용하면 된다.

(예를들어 01/03(start_date)부터 01/07(now)까지 DAG RUN을 실행했고, start_date 이전인 01/01부터 01/02 기간동안 DAG RUN을 실행하고자 한다면, Backfilling mechanism을 위한 명령을 사용하면 된다)

1

with DAG('my_dag', start_date=datetime(2022, 1, 1), schedule_interval='@daily', catchup=False) as dag:

이렇게 catchup=False로 설정값을 바꿔주면, non-triggered DAG RUN이 실행되게 된다. 이 mechanism은 과거에 non-triggered DAG RUN을 자동으로 재실행할때 사용된다.

Executor

이전 포스팅에서도 다뤘던 내용이지만, Executor는 이름 자체는 Task를 실행할 것 같지만, Task를 실행하지 않는다. 단지 tasks를 시스템에서 어떻게 실행할 것인가에 대해 정의한다.

Executor에는 다양한 종류가 있는데, local executors와 remote executors가 있다. local executor는 여러 개의 task를 single machine에서 실행을 하고, sequential executor는 single machine에서 한 번에 하나의 task를 실행할때 사용된다.

remote executor에는 Celery executor가 있는데, tasks를 multiple machine, 그리고 salary cluster에서 실행한다. K8s executor는 multiple machine에서 K8s cluster의 multiple pods에서 multiple tasks를 실행한다.

Executor의 변경은 Airflow의 환경설정 파일에서 executor parameter를 변경함으로써 적용할 수 있다. (사용되는 executor에 따라 변경해야 되는 별도의 환경설정 요소가 있다)

venv 명령어

1
2
3

$python -m venv venv # venv이름으로 가상환경 생성
$source venv/bin/activate # 가상환경 활성화 시키기
$deactivate # 가상환경 나가기

pip 명령어

pip는 Python의 패키지 매니저로, 외부 패키지나 라이브러리, 프레임워크를 설치하고 관리할 수 있도록 도와준다.

1
2
3
4
5
6

$pip install pip --upgrade # pip upgrade
$pip install "package~=3.0.0" #3.0.0 version의 패키지를 설치
$pip install [package] # package 설치
$pip uninstall [package] # package 삭제
$pip --version # 설치된 pip version을 확인할 수 있다.
$pip freeze # 설치된 패키지를 확인할 수 있다.

설치된 페키지를 text로 보내고 설치하기(협업)

1
2

$pip freeze > requirements.txt # requirements.txt 파일에 설치된 패키지 리스트를 파일로 뽑아내기
$pip install -r requirements.txt # requirements.txt파일에 기록된 패키지를 설치

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15

# python version: 3.8.1

autopep8==1.6.0
click==8.1.3
flake8==5.0.4
importlib-metadata==4.12.0
itsdangerous==2.1.2
Jinja2==3.1.2
MarkupSafe==2.1.1
mccabe==0.7.0
pycodestyle==2.9.1
pyflakes==2.5.0
toml==0.10.2
Werkzeug==2.2.1
zipp==3.8.1

CPU 바운드, I/O 바운드, Blocking

바운드

바운드란 장애물에 막혀서 실행이 되지 않는 상태를 말한다.

CPU 바운드

프로그램이 실행될 때 실행속도가 CPU 속도에 의해 제한되는 것을 말하며, 복잡한 수학 수식을 계산하는 경우, CPU의 연산 작업에 의해 프로그램이 실행될때 실행속도가 느려지거나 멈춰있는 되는 현상이 발생하게 되는데, 이를 CPU 바운드라고 한다.

I/O 바운드

프로그램이 실행될 때 실행속도가 I/O에 의해 제한되는 것을 말하며, 프로그램에서 사용자의 입력을 기다리기 위해 프로그램이 멈춰있는 경우가 발생하는데, 이를 I/0 바운드라고 한다.

Network I/O 바운드

사용자로부터 입력을 기다리기 위해 프로그램이 멈추는 것이 아닌, 외부 서버에 요청을 하여 응답을 기다리는 경우에도 프롤그램이 멈춰있는 현상이 발생하는데, 이를 Network I/O 바운드라고 한다.

Blocking

바운드에 의해 코드가 멈추게 되는 현상이 일어나는 것을 블로킹이라고 한다.

Apache Airflow UI 구성

Apache Airflow UI에서 DAGs 리스트를 보면, 현재 Apache Airflow에 포함되어있는 DAG의 목록이 출력된다. 리스트 중 하나의 아이템을 살펴보면, DAG의 이름의 좌측에 해당 DAG를 Pause/Unpause 할 수 있는 토글 버튼이 있고, 이름의 아래에는 data pipeline이나 팀 또는 기능별로 묶어서 관리할 수 있도록 태그가 표시되어있다. (단, tag에 따라 permission을 부여할 수 없다)

그 외에 Owner에 대한 정보와 Runs 칼럼에서는 queued, success, running, failed 상태별로 현재 DAG이 실행 상태를 확인할 수 있다. Last Run 및 Next Run 항목을 통해서는 DAG가 언제 마지막으로 실행이 되었고, 그 다음 실행은 언제 되는지에 대해 확인할 수 있다.

Recent Tasks에서는 총 15개의 상태 정보로 나뉘어 활성화된 DAG의 Task들의 실행 상태에 대해서 확인을 할 수 있다. (none, removed, scheduled, queued, running, success, shutdown, ...)

Actions에서는 강제로 Trigger 시키거나 DAG를 삭제할 수 있는데, DAG를 삭제한다는 의미가 DAG 자체를 삭제하는 것이 아닌, Meta store에서 DAG RUN Instance를 삭제한다는 것을 의미한다.

Apache Airflow DAG item 상세보기

DAG의 이름을 클릭하면, Grid를 통해 실행한 DAG들의 상태 정보에 대해서 모니터링 할 수 있으며, Graph View를 통해서는 DAG의 각 Tasks가 어떤 Tasks를 dependencies로 가지고 있는지에 대해서 구조적으로 확인을 할 수 있다.

Landing Times에서는 DAG에서의 모든 작업들이 scheduled 상태에서 completion으로 완료되는데 얼마나 걸렸는지에 대한 정보를 시각화된 그래프로 확인할 수 있다. 만약 시간이 오래걸린다면, 걸리는 시간을 줄일 수 있도록 별도의 대응이 필요하다.

Calendar view에서는 각 각의 네모칸이 특정 DAG의 상태 다이어그램 정보를 종합해서 보여준다. 특정 DAG가 특정 날짜에 문제가 생긴다면, 빨간색으로 칸이 표시되며, 성공적으로 DAG가 시행이 되었다면 해당 일자에 초록색으로 포시된다. 점으로 표시된 칸은 얼마나 많은 다이어그램이 해당 날에 계획되어있는지에 대한 정보를 제공한다. 따라서 이러한 모니터링 정보를 기반으로 어떤 날에 데이터 파이프라인을 수정해서 문제를 해결해야되는지 알 수 있다.

Gantt view에서는 DAG의 특정 Task가 완료되는데에 얼마나 시간이 걸렸는지와 pipeline에서 bottleneck이 발생했는지에 대한 전반적인 overview를 제공한다. 상대적으로 긴 직사각형은 그만큼 Task를 실행하는데 시간이 걸렸다는 의미이고, 오래걸린 작업에 대해서는 어떻게 하면 작업이 완료되는데 걸리는 시간을 단축시킬 수 있는지에 대한 고민이 필요하다.

직사각형이 overlapping되었다는 것은 복수 개의 Task를 동시에 실행할 수 있다는 것을 의미한다. (DAG parallelism)

Code view에서는 데이터 파이프라인의 코드에 접근할 수 있는데, 이를 통해서 적용한 수정사항이 DAG에 제대로 적용이 되었는지 확인을 할 수 있다.

이번 포스팅에서는 Apache Airflow의 기본 개념과 사용에 대해서 정리해보려고 한다.
이전에 AWS의 EventBridge라는 서비스를 사용해서 셍성한 Lambda 함수를 일정 주기의 시간동안 정기적으로 실행되도록 스케줄링해서 실습한 적이 있는데, Task를 좀 더 복잡한 구조로 스케줄링하기 위해서는 Apache Airflow를 활용하는 것이 좀 더 효율적이라고 오프라인 수업에서 배워서, 한 번 Apache Airflow를 사용해서 데이터 파이프라인을 구축해보고자 학습을 하게 되었다. 그리고 지원하고자 하는 기업에서 Apache Airflow를 업무에서 도입을 하여 사용하고 있기 때문에 좀 더 잘 이해하고 사용해보려고 한다.

Apache Airflow를 사용하는 이유

데이터 파이프라인 구축에 있어, 데이터를 추출하는 Extract, 데이터를 적재하는 Load, 데이터를 변환하는 Transform 과정을 거친다. 데이터 추출은 API를 통해서 하기도 하며, Load는 Snowflake와 같은 data warehousing, data lake와 같은 Single platform을 제공하는 SaaS를 활용하기도 한다. 그리고 Transform은 Dbt와 같은 분석을 위한 데이터 웨어하우스에 적재된 데이터를 간단한 SELECT 문 작성을 통해 변환을 할 수도 있다.

만약 Extract, Load, Transform 단계에서 예기치 못한 에러가 발생한다면, 그리고 데이터 파이프라인이 한 개가 아닌 100개 이상이라면 어떨까?

관리하기가 많이 어려워진다. 이러한 이유로 인해 Airflow를 사용해서 종합적인 파이프라인의 관리가 필요하다.

DAG(Directed Acyclic Graph)

DAG는 방향성 비순환 그래프로, Airfow의 주요 컨셉이며, 복수 개의 Task를 모아서 어떻게 실행이 되어야 하는지에 대한 종속성과 관계에 따라 구조화 시키는 것을 말한다.

Operator

Operator는 Task이며, 실행이 되면 Task instance가 생성이 되며, Operator의 예로는 Action operator, Transfer operator, Sensor operator가 있다.

Action operator의 예로는 Python operator, Bash operator가 있는데, Python operator는 Python function을 실행시키며, Bash operator는 Bash command를 실행시키는 역할을 한다. 그리고 Transfer operator의 예로는 MySQL의 데이터를 RedShift로 이전하는 작업이 있으며, Sensor operator의 예로는 특정 이벤트가 발생하면 다음 Step으로 넘어가도록 하는 작업이 있다.

Apache Airflow의 요소

Apache Airflow에는 Webserver, Meta store, Scheduler, Executor, Queue, Worker가 있다.

Executor는 직접적으로 Task를 실행하지 않으며, K8S 클러스터는 K8S Executor를 사용하고, Celery 클러스터는 Celery Executor를 사용한다. 여기서 Celery는 multiple machine에서의 multiple tasks를 실행하기 위한 Python 프레임워크이다.

Queue는 주어진 Task를 보장된 순서로 실행시키기 위해 존재한다.

Worker는 Task가 효과적으로 실행될 수 있도록 하는 역할을 하며, Worker가 없다면, sub processes 혹은 K8S를 사용하는 경우, Path가 주어진다.

Apache Airflow의 Architecture

Apache Airflow의 Architecture로는 단일 노드로 구성된 One Node Architecture, 그리고 복수 개의 노드들로 구성된 Multi Node Architecture가 있다.