눈가락★

Docker Compose 는 다수의 "docker build" 와 다수의 "docker run" 명령어를 대체하는 도구(orchestration 명령 모음)

Docker compose 툴로 다수의 container 들을 한 번에 손쉽게 관리할 수 있음

하나의 Docker compose 를 사용하여 모든 container 들을 빌드, 실행, 및 중지/중단 할 수 있음 

Docker compose 는 Dockerfile 을 대체하는 용도가 아니며,

역시 image 혹은 container 대체하는 용도가 아님 

Docker compose 는 여러 host 위에서 동작하는 다중 컨테이너들을 관리하는 데 적합하지 않음

단지 하나의 동일한 host 위에서 동작하는 다중 컨테이너들을 관리할 수 있음

Docker compose 에서 관리하는 Container 들은 기본적으로 -d --rm 옵션을 갖게 됨

Docker compose 에서 관리하는 Container 들은 모두 하나의 네트워크로 묶이게 됨

즉, 하나의 동일한 Docker compose 파일에 정의된 모든 서비스(Containers)는

(별다른 네트워크 설정을 하지 않아도) 동일한 네트워크의 일부가 되어 서로 통신이 가능함

Docker compose 구성 내에서 Container 는 "Service" 라는 이름으로 불림

이 Service 에 어떤 port 를 열것인지 어떤 네트워크를 붙일 것인지

볼륨을 추가 할 것인지 환경변수를 추가할 것인지 등등을 docker compose 내에 정의할 수 있음

Docker compose 구성 파일은 "docker-compose.yml" 혹은 "docker-compose.yaml" 라는 이름으로 만듦

docker-compose.yml 의 포맷은 다음 예제(왼쪽)와 같음

- version : Docker comose 사양의 버전을 지정

  (앱이나 docker-compose.yml 파일의 버전이 아님) 

  version 은 해당 docker-compose yml 파일에서 사용할 수 있는 기능에 영향을 미친다고 함 

  계속 업데이트되는 Docker compose 중에서 어느 기능을 사용할 지 version 으로 고정시켜주는 거라고 함

  (근데 최근 문서를 보니 version 은 안 쓰는 것 같기도..? 일단 강의 내용만 따라가보자)

- services : Container 를 지정

  위 예제에서는 총 세 개의 containers 를 생성함

  mongodb_container, container2, container3

  각 container 가 어떤 이미지를 기반으로 생성될지, 어떤 네트워크에 붙을지 등을 정의하면 됨

- image : 생성될 container 의 기반으로 사용할 이미지

- volumes : container 볼륨 설정

  volume 여러개 추가 가능

  ro 등의 옵션을 추가할 수 있음 (예를 들어 -v data:/data/db:ro)

- environment : container 내에서 사용할 환경 변수 지정

  위 예제처럼 

      environment:
      - MONGO_INITDB_ROOT_USERNAME=max

  이렇게 지정해도 되고 아래처럼 지정해도 됨

      environment:
        MONGO_INITDB_ROOT_USERNAME: max

  혹은 environment 대신 env_file 을 사용하여, 환경변수 파일(mongo.env)을 넣어 지정할 수 있음

      env_file:
        - ./env/mongo.env

  환경변수 파일(mongo.env) 내용은 다음과 같아야 함

- networks : container 에 추가 network 를 붙임

  위에서 설명한 것처럼, Docker compose 가 관리하는 모든 서비스(Containers) 는

  자동으로 하나의 기본 네트워크에 묶이게 되지만

  networks 명령어를 추가하면 해당 Container 에는 추가 네트워크가 붙게됨

  위 예제에서 mongodb_container 는 기본 네트워크와 goals-net 네트워크 두 군데에 붙게 됨

- (마지막줄에 있는) volumes : container 에 적용한 volume 의 이름을 넣어주는 곳

  위의 예제에서 mongodb_container 에 volume 이름이 "data" 이므로

  마지막 volumes 명령어에 "data" 를 넣어주었음.

  이렇게 따로 사용되는 volume 이름을 넣어줌으로써, Docker 가 명명된 volume 이름을 인식할 수 있다고 함

  (익명 volume 이나 path 를 직접 지정한 bind mount volume 은 마지막 volumes 명령어로 넣어주지 않아도 됨)

  추가로, 다른 서비스에서 동일한 volume 이름을 사용하면, 그 volume 이 공유된다고 함

  (서로 다른 Container 들이 host 머신 상의 동일한 volume 을 동시에 사용 가능)

docker-compose.yml 파일을 생성한 후, 동일한 위치에서 아래 명령어 실행

그러면 docker-compose.yml 에 설정한 모든 서비스가 시작됨

(-d 옵션이 없으면 Docker compose 자체가 attached 모드로 시작됨)

docker-compose.yml 실행을 중지하려면 아래 명령어 실행

위 명령어로 volume 은 제거되지 않는데, 

만약 제거하고 싶다면 -v 옵션을 붙임 (docker-compose down -v)

docker-compose.yml 의 다른 예제도 확인해봄

- build : service 를 생성할 이미지를 직접 빌드해서 사용함

  단순히 build: ./backend 이런 식으로 짧게 쓸 수 있고 (이 경우, Dockerfile 이름은 반드시 "Dockerfile" 이어야 함)

  context: 와 dockerfile: 등 을 추가하여 좀 더 자세하게 명시할 수 있음

  context: 는 Dockerfile 이 존재하는 위치이며, build 했을 때 기준이 되는 위치임

  위의 예제에서 ./backend 위치를 기준으로 build 를 진행하며 COPY 등의 기준 위치가 됨

  dockerfile: 은 Dockerfile의 이름임

  (위의 예제에서 Dockerfile 의 이름은 default 인 "Dockerfile" 이지만, 내 맘대로 지은 "eyeballsfile" 도 될 수 있겠지)

  args: 를 통해 Dockerfile 내부에서만 사용 가능한 상수값을 지정할 수 있음

  Dockerfile 생성할 때 넣은 ARG 명령 사용한 것과 동일한 부분임

- ports : service 에서 노출하고 싶은 포트 지정 가능

  위 예제의 '80:80' 은 service 포트(앞 80)와 이미지 내부 포트(뒤 80)가 연결됨

  '123:443' 은 service 포트(앞 123)와 이미지 내부 포트(뒤 443)가 연결됨

  service 사용자는 80 혹은 123 포트를 사용하여 이미지 내부 80 혹은 443 으로 접근 가능

- volumes : 위 예제와 같이 상대 경로에 해당하는 위치를 volume 으로 mount 가능

  이렇게 경로를 이용하여 mount 한 경우, 마지막줄에 volume 의 이름을 넣어주지 않아도 괜찮음

- stdin_open tty : 사용자가 해당 service 와의 상호작용(interactive)을 필요로 하는 경우 true

- depends_on : docker compose 를 통해 동시에 여러 container(service) 들을 생성할 때

  이미 실행되고 있는 다른 container 에 의존성을 설정할 수 있음

  예를 들어 A container 가 먼저 실행되고 그 후에 B container 를 실행할 수 있도록 순서를 정할 수 있다는 의미

  위의 예제에선, 'mongodb_container' 와 'A_container' 두 개의 container 가

  먼저 실행 된 이후에 'backend_container' container 가 나중에 실행될 것임

docker-compose.yml 을 통해 생성한 container 들을 docker ps 를 통해 확인하면

우리가 지정한 이름과 조금 다른 것을 확인할 수 있음

위에 docker-compose.yml 예제를 통해 mongodb_container, backend_container 두 가지 container 를 생성 후

docker ps 명령어를 확인해보면 아래와 같은 이름이 보임

- docker-practice_backend_container_1

- docker-practice_mongodb_container_1

container 이름은 service 이름에 접두사와 접미사가 붙은 형태이며

접두사 "docker-practice" 는 프로젝트 폴더 이름,

접미사 "1" 은 증가하는 숫자임

위와 같은 이름 대신, 직접 container 이름을 지정하려면 container_name 명령어를 사용하면 됨

이를테면

container 이름에 접두사, 접미사가 붙긴 했지만,

docker compose 가 생성한 기본 network 에서는 우리가 지정한 service 의 이름으로 container 끼리 통신이 가능함

즉, backend_container 에서 "mongodb_container" 이름으로 mongodb_container 에 접근 가능하다는 것임

docker container 의 기반이 되는 각 이미지는

docker run 명령어 실행시 실행되는 기본 프로세스(pid 1번 프로세스)가 있음

이 기본 프로세스를 이미지 변경 없이(즉, Dockerfile 의 CMD, ENTRYPOINT 추가를 통한 업데이트 없이) 바꿀 수 있음

예를 들어 다음과 같이 실행한 경우

- docker run -it node

node 라는 이미지는 기본 프로세스가 "node" 이기 때문에 node 가 실행됨

하지만 다음과 같이 실행한 경우

- docker run -it node npm init

"node" 대신 "npm init" 이라는 명령에 의한 프로세스가 기본 프로세스로 실행됨

따라서 npm init 명령에 의한 프로세스가 해당 컨테이너의 생명주기에 관여함

CMD 가 존재하는 Dockerfile 로 빌드한 이미지를 docker run 으로 실행할 때

뒤에 추가 명령어를 붙여서, 기본 프로세스를 덮어쓰게되면

CMD 로 지정한 실행명령어가 사라지고 run 뒤에 붙인 추가 명령어가 대신 실행됨

예를 들어보자

docker run 을 실행하면 CMD 에 의해 "hello eyeballs!" 가 출력되어야 하지만

run 뒤에 추가한 echo "bye" 명령어에 의해 "bye" 가 대신 출력됨

CMD 와 비슷한 역할을 하는 ENTRYPOINT 는 조금 다르게 동작함

ENTRYPOINT 가 존재하는 Dockerfile 로 빌드한 이미지를 docker run 으로 실행할 때 뒤에 추가 명령어를 붙이면

ENTRYPOINT 로 지정한 실행명령어의 뒤에 추가한 명령어가 붙어 같이 실행됨

예를 들어보자

docker run 명령어 뒤에 "eyeballs" 를 추가해 줌

docker run 을 실행하면 ENTRYPOINT 에 의해 실행되는 echo hello 뒤에 eyeballs 가 붙어서

echo hello eyeballs 라는 명령어가 기본 프로세스가 되었음

따라서 "hello eyeballs" 가 출력됨

비슷한 논리로, Docker Compose 를 통해 기본 프로세스를 덮어쓸 수 있음

다음과 같은 Dockerfile 과 docker-compose.yml 이 존재한다고 하자

위와 같이 docker compose run 명령어 뒤에

service 이름과 추가할 명령어를 넣으면

기본 프로세스에 영향을 미칠 수 있음

위 예제에서는 Dockerfile 이 ENTRYPOINT 명령어를 실행하기 때문에

명령어가 뒤에 추가되어 hello eyeballs 가 출력되었음

docker compose run --rm my_service 처럼 --rm 옵션을 추가해주면

docker compose run 명령어를 통해 실행된 컨테이너가 stop 될 때 컨테이너 자체를 삭제할 수 있음

docker-compose.yml 에 설정된 services 중 내가 원하는 service 만 실행할 수 있음

아래와 같은 docker-compose.yml 에서 bb_svc, cc_svc 만 실행하고 싶다고 하자

아래와 같이 docker compose up 명령어 뒤에 service 이름을 넣어주면

지명된 services 만 실행됨 

Dockerfile 에서 RUN 을 통해 실행되는 모든 명령은

도커 컨테이너 및 이미지의 작업 디렉토리에서 실행되는데

여기서 말하는 작업 디렉토리는 컨테이너 파일 시스템의 루트 폴더

예를 들어 아래와 같은 Dockerfile 이 존재한다면

node 컨테이너의 / 위치에서 npm install 명령을 실행하게 됨

특정 위치에서 RUN 명령어를 실행하기 위해서는 WORKDIR 를 사용

이를 통해 작업 디렉토리를 바꿀 수 있음

예를 들어

COPY . ./ 는 "현재 Dockefile 이 위치해있는 곳의 모든 파일,디렉터리들" 을

"컨테이너 내 작업 디렉터리" 에 복사하는 의미의 명령어

RUN : Dockefile 을 통해 이미지를 생성할 때 수행 할 작업을 지시

ENTRYPOINT, CMD : Dockerfile 을 통해 이미지 생성할 때 수행되지 않고,

이미지가 만들어진 후 그 이미지로 컨테이너가 처음 시작될 때 컨테이너 내부에서 수행 할 작업을 지시

CMD 명령어는 특이하게 명령을 배열에 담아줘야 함

예를 들어 echo hello 라는 명령을 넣고싶다면

EXPOSE 명령어를 통해, 도커 컨테이너 실행시 -p 옵션을 통해 연결 가능한 (컨테이너 내) 포트를 문서화 할 수 있음

Dockerfile 을 만든 후 이미지로 만드는 방법은 다음과 같음

ENTRYPOINT 와 CMD 를 자세히 알아봄.

위에 기술했듯이 ENTRYPOINT, CMD 둘 다

컨테이너가 처음 시작될 때(docker run 명령어로 실행시) 컨테이너 내부에서 수행 할 작업을 지시하는 명령어임

CMD 하나만 존재하는 Dockerfile 을 확인해보자

위와 같은 Dockerfile 로 이미지를 만들어서

docker run 명령어로 실행하면

CMD 에 넣은 명령어가 실행되는 것을 볼 수 있음

이렇게 CMD 로 실행된 프로세스(echo)는 pid 1 번으로 실행되어 컨테이너의 생명 주기에 관여함

1번 프로세스가 살아있어야 컨테이너가 계속 살아있게 되고

만약 1번 프로세스가 죽게되면 컨테이너도 같이 stop 됨

실제로 docker stop 명령어가 컨테이너를 stop 하는 방법은,

pid 1번 프로세스에 SIG TERM 시그널(프로세스 종료 신호)을 보내는 거라고 함

위 Dockerfile 로 만든 컨테이너도 run 명령어를 날리자마자 echo 출력하고 곧바로 stop 되었다.

반대로,ENTRYPOINT 가 사용된 Dockerfile 은 어떻게 동작하는지 보자

ENTRYPOINT 역시 docker run 명령어로 실행을 통해 곧바로 실행되는 것을 볼 수 있음

CMD 와 마찬가지로 ENTRYPOINT로 실행된 프로세스(echo)는 pid 1 번으로 실행되어 컨테이너의 생명 주기에 관여하게 됨

(docker run 명령어 이후 echo 출력하고 곧바로 stop 됨)

그럼 CMD 와 ENTRYPOINT 차이는?

ENTRYPOINT 명령어가 먼저 오고, CMD 가 ENTRYPOINT 명령의 파라미터를 넣어주는 역할을 한다고 함

docker run 을 통해 실행되는, 생명주기에 관여하게되는 pid 1번 프로세스는 ENTRYPOINT 의 명령어로 실행되고

CMD 명령어는 ENTRYPOINT 명령어에 이어 붙이는 파라미터, arguments 들을 넣어주는 역할

예를 들어보자

docker run 을 실행하면

top -b -c 명령어가 실행됨

docker inspect 를 통해 확인해보면 다음과 같이 CMD, ENTRYPOINT 명령어가 들어있는 것을 볼 수 있음

추가) docker run 실행시 --entrypoint 옵션으로 pid 1번 프로세스 실행 변경이 가능함

참고 : https://asung123456.tistory.com/51

docker run 명령어는 attatched 모드로 컨테이너를 실행하고

docker start 혹은 restart 명령어는 detached 모드로 컨테이너를 실행

attatched 모드는 포그라운드에서 동작하면서 컨테이너 내의 출력 결과가 계속 출력되는 모드

detached 모드는 백그라우드에서 동작하며 사용자와 인터랙션이 없는 모드

docker run 명령어로 detached 모드를 사용하려면 -d 옵션을 붙이면 됨

반대로 docker start 명령어로 attached 모드를 사용하려면 -a 옵션을 붙이면 됨

detached 모드로 실행중인 컨테이너에 attached 모드로 붙으려면 docker attach 명령어 사용하면 됨

(docker run 실행했을 때 처럼 attached 모드로 붙을 수 있음)

docker container 의 로그를 지속적으로 보고 싶다면 -f 옵션 추가

도커가 run 되면 echo program(사용자 입력을 받고 그대로 출력하는 ) 파이썬 스크립트를 실행하는 이미지(echo-python)가 있다고 하자

이 이미지로 run 하면

사용자 입력을 받아야하는데 받지 못해서 에러가 남

사용자 입력을 받을 수 있도록 하려면 -it 옵션을 추가해야 함

  -i, --interactive                    Keep STDIN open even if not attached
  -t, --tty                            Allocate a pseudo-TTY (teletypewriter)

내가 흔히 사용하는, docker run -d image bash 와 docker exec -it container bash 두 단계를 거치지 않아도

docker 이미지 내 CMD 명령어를 통해 처음 실행하는 프로그램을 곧바로 사용 가능

이런 컨테이너가 stop 된 이후에 start 하게 되면

기본적으로 detached 모드가 되니 상호작용이 불가능

따라서 start 할 때 -a 옵션을 줘서 attached 모드로 실행하고, -i 옵션을 줘서 interactive 활성화 해야 함

현재 컨테이너로 사용되지 않는, 모든 이미지를 지우는 명령어

  prune       Remove unused images

현재 stop 상태인 모든 container 를 지우는 명령어

  prune       Remove all stopped containers

다운받은 Docker Image 의 상세 정보, 예를 들어

어떤 port 를 오픈할 수 있는지, 어떤 이미지를 base 로 삼고 있는지,

어떤 CMD 명령어를 실행하는지, entrypoint가 무엇인지 등을 알고 싶다면 inspect 명령어 사용

Docker 컨테이너가 삭제되어도, 동일한 이미지로 다시 컨테이너를 띄웠을 때

데이터를 유지할 수 있는 방법은 volume 옵션을 사용하는 것

내가 알고있는 volume 은 host 의 path 와 연결하는 것인데,

docker 자체에서 알아서 path 를 정해서 host 어딘가에 데이터를 저장하는 방법을 제공할 수도 있음

예를 들어

위와 같은 volume 옵션을 준다면,

내 local host 머신의 /home/eye/docker 내부의 모든 파일, 디렉터리들이

docker container 내부의 /app 위치에 동기화 될 것이며

내가 host 머신에서 해당 파일, 디렉터리들에 직접 접근이 가능함

하지만 아래처럼 local path 가 아닌, volume 이름만 준다면

docker 가 내 local host 머신 어딘가에 myvolume 에 해당하는 위치를 지정해두고

docker container 내부의 /app 에 저장되는 파일, 디렉터리들을 myvolume 위치에 동기화 시킬 것임

나는 host 머신에서 myvolume 위치에 접근이 불가능함

만약 아래처럼 이름조차 주지 않는다고 하더라도

docker 가 내 local host 머신 어딘가에 익명의 이름으로 path 를 지정해두고

docker container 내부의 /app 에 저장되는 파일, 디렉터리들을 동기화시킴

그래서 container 가 죽어도 파일/디렉터리들을 유지할 수 있음

위와 같이 내가 접근이 불가능한, 이름으로 volume 을 주고 그 목록을 보고자 한다면 아래 명령어 사용

docker volume 을 삭제하려면 아래 명령어 사용

당연한 이야기지만 아래 명령어로 volume 을 삭제하면,

volume path 에 저장한 데이터들도 모두 삭제됨

특정 이름/익명의 이름으로 추가된 volume 은 container 가 관리하게 되지만

내 local 머신의 path 와 container 내 path 를 잇는 volume 은 container 가 관리하지 않아서

docker volume ls 명령어에 나타나지 않음

volume 을 사용할 때 주의할 점은,

Dockerfile RUN 명령어를 통해 실행되고 저장된 어떤 데이터를

Docker volume 으로 덮어써버릴 수 있다는 것임

예를 들어

위와 같은 Dockerfile 은 npm install 명령어를 통해 /app 에 설치 데이터를 저장함 (정확히는 /app/node_modules 에 저장됨)

아래 명령어로 Docker run 을 하게되면

Container 내부 /app 위치에 저장되어있던 npm 설치 데이터들이 모두 덮어쓰여지게 됨

이런 경우를 위해, 덮어쓰지 말아야 하는 디테일한 path(/app/node_modules) 는 익명 volume 으로 처리하는 옵션을 추가

위와 같이 지정해주면 docker container 내부의 /app/node_modules 가 첫번째 볼륨에 의해 덮어쓰여지지 않게 됨

docker volume 을 지정하면,

local 머신과 container 내부가 서로  공유하는 path 가 생기고

이 path 에 존재하는 파일들에 서로 read 와 write 가 가능함

local 머신에서도 파일을 읽고 쓸 수 있고

container 내부에서도 파일을 읽고 쓸 수 있다는 말임

그런데 나는 container 내부에서 파일을 읽기만 하고 쓰지는 못하도록 하고 싶음

즉, local 머신에서만 쓸 수 있도록 하고 싶음

그럴 땐 volume 옵션에 read only 성격을 추가하면 됨

위와 같이 ro 옵션을 주면, container 내부에서는 /app 및 /app 이하 모든 path 에서 write 가 불가능하게 됨

volume 옵션이 두 가지 있다면,

좀 더 깊이있는 path 를 갖는 volume 옵션의 우선순위가 더 높음

예를 들어 다음과 같은 옵션이 있을 때

/app:ro 보다 /app/modules 가 좀 더 깊이있기 때문에

/app/modules 의 우선순위가 높음

즉, container 에서는 /app path 에 있는 파일에 write 가 불가능하지만

/app/modules path 에 있는 파일에는 write 가 가능하다는 말임

docker container 가 관리하는 volume 의 자세한 내용을 보고 싶다면

(언제 생성되었고 이름은 무엇이고 local 머신 어디에 저장되어있는지... 등등) 

위 명령어로 mountpoint 가 나타나며

이는 docker container 의 가상머신 상의 경로라서

local 머신에서 찾을 수 없음

현재 사용되지 않는 모든 local volume 을 지우는 명령어

prune       Remove unused local volumes

Dockerfile 에서 COPY 명령어로 데이터를 container 내부로 복사할 때

복사하지 말아야 할 리스트인 .dockerignore 를 만들어서 복사하지 않도록 만들 수 있음

docker container 내에 환경변수를 넣어줄 수 있음

여기서 말하는 환경변수는 linux 에서 사용하는 환경변수 그거임

예를 들어 아래와 같은 Dockerfile 을 docker run 해보면

이렇게 container 내부 환경변수를 미리 지정할 수 있음

환경변수를 지정하는 방법은 총 3가지

1. Dockerfile 에 ENV 명령어로 지정

  위 예제대로 넣어줄 수 있음

2. docker run 명령어에 --env 혹은 -e 옵션으로 지정

  위 예제처럼 MYNAME 을 넣으려면 아래처럼 옵션 추가

3. docker run 명령어에 --env-file 로 환경변수가 저장된 파일를 지정

만약 1번 방법과 2번 방법이 동시에 사용된다면?

2번 방법으로 들어간 환경변수가 더 높은 우선순위를 갖게 됨

만약 2번 방법으로 환경변수를 넣어줘야 하는데 넣지 못했다면,

1번 방법으로 들어간 환경변수가 대신 사용됨

즉, 1번 방법으로 들어간 환경변수는 default 값이 되는 것임

강의에서 "ENV MYNAME eyeballs" 처럼 설정했다가, 환경 변수가 default 로 적용되지 않아서

"ENV MYNAME=eyeballs" 처럼 등호를 넣어 다시 설정하니 되는 것을 보았음

1번 방법이 default 로써 사용되려면 등호를 넣어서 설정해야 하나 봄

ARG 명령어는 Dockerfile 내부에서만 사용 가능한 상수값

아래처럼 ARG 에 설정한 값은 ENV 등에서 사용이 가능함

위와 같이 Dockerfile 내부에서 지정할 수 있고,

docker run 명령어에 옵션을 주어 지정도 가능

ARG 가 지정되고나면 값은 고정되어버림

위와 같이 옵션을 통해 값을 지정하면, 이미지는 동일하나 ARG 값이 다른 여러 container 를 만들 수 있음

참고로 ARG 값은 CMD 명령어에서 사용 불가능함

ARG 명령어와 ARG 로 설정한 값을 사용하는 명령어는

Dockerfile 의 마지막에 넣어두는 게 좋음

ARG 값이 바뀌면 이미지를 빌드할 때 (업데이트가 된) ARG 명령어 이후의 모든 명령어를 다시 빌드하게 되기 때문

빌드를 다시 하지 않아도 되는 명령어들까지 빌드하게 되므로 리소스 및 시간이 낭비되기 때문

Docker container 내부에서, WWW (외부 인터넷) 에 존재하는 웹사이트 접근이나 API 요청하기 위해

따로 설정해야 하는 것은 없음. 그냥 가능함

Docker container 내부에서, container 가 실행되는 host 머신에 접근하려면

(localhost 가 아닌) host.docker.internal 로 접근하면 됨.

예를 들어 host 머신 위에 mongodb 가 실행중이고

Docker container 내부에서 host 머신의 mongodb 에 접근하려면 아래와 같은 방식으로 접근 (수도코드)

Docker container 내부에서 host 머신으로 ping 을 날릴때도 이와 비슷함

Docker container 간 네트워크 연결이 가능함

예를 들어 AA container 와 BB container 가 존재하고

AA 가 BB 에게 ping 을 날리는 상황을 가정하자

AA container 내부에서 ping 을 실행하려면

BB container 의 ip 주소를 알아야 함

BB container 의 ip 주소는 docker container inspect BB 명령어 결과에서 볼 수 있음

결과중에 NetworkSettings.IPAddress 부분을 보면 BB container 의 ip 주소가 나타남

이렇게 얻은 ip 주소를 이용하여 AA container 는 ping 을 날릴 수 있음

위와 같이 ip 주소를 직접 얻어 사용하면

ip 주소를 사용하는 코드에서 하드코딩을 하게 되고

또 ip 주소가 바뀌면 하드코딩 또한 바뀌고 이미지 빌드도 다시 해야하고 실행도 다시 해야하기 때문에 아주 불편함

따라서, container 간 통신을 위해 ip 주소를 사용하는 대신,

하나의 network 로 container 들을 묶는 기술을 사용해야 함 (바로 아래서 설명)

하나의 docker network 를 만들고

그 network 에 container 들을 연결시켜두면

container 들끼리 자기들의 이름으로 통신이 가능하게 됨

아래 명령어를 통해 network 를 하나 만듦

network 의 이름은 간단하게 mynetwork 라고 하자

만든 network 는 아래 명령어로 확인이 가능

각 container 를 실행할 때 내가 만든 mynetwork 에 연결하여 실행

예를 들어 AA container, BB container 를 실행할 때 아래처럼 --network 옵션을 추가하여 연결

그럼 AA container 내부에서 BB container 의 이름(BB)으로 ping 을 날릴 수 있게 됨

AA container 에서 Mongodb container 에 연결하려면 docker run 실행시 -p 옵션으로 포트를 열어야할까?

아님. -p 옵션은 host 머신에서 달라붙기 위해 오픈하는 포트 번호를 넣는 옵션일 뿐

container 간 연결에서는 -p 옵션을 넣지 않아도 container 간 포트가 열려있음

어떤 프로그램, 툴 의 공식 이미지는 '사용 방법'이 다 정해져서 나옴

예를 들어 mongodb 공식 이미지는 "mongo"이며

https://hub.docker.com/_/mongo 에서 확인할 수 있음

공식 이미지는, 만들어질 때 내부에서 프로그램을 어떻게 운용할지 정해둠

즉, pid 1번 프로세스는 무엇이며

로그는 어느 dir 에 저장하며

기본 port 는 무엇이며

환경변수에 따라 어떤 설정값이 추가되는지 등

그리고 그 '사용 방법'은 대개 docker image 에 안내되어있음

예를 들어 mongodb 공식 이미지의 경우

mongodb 프로세스가 데이터를 저장하는 dir 위치를 /data/db  로 설정해두었고

mongodb 의 root 사용자를 정의하려면

MONGO_INITDB_ROOT_USERNAME,  MONGO_INITDB_ROOT_PASSWORD 라는

두 가지 환경 변수를 사용해야 한다는 내용이 이미 image 내에 포함되어 있음

따라서 우리는 공식 이미지를 사용할 때, '사용 방법'에 따라 사용하면 됨

(우리가 '사용 방법'을 직접 추리할 수는 없는 노릇)

Docker Container 는 리소스(memory, cpu, io bps) 제한이 없음

호스트의 리소스를 제한 없이 사용함

따라서 Container 에서 리소스를 많이 사용하는 만큼

호스트 머신에서 동작하는 다른 프로세스에 영향이 끼침

심지어 Container 의, Container 에 의해 Out Of Memory 도 발생할 수 있음

Container 의 Memory 및 CPU 제한을 확인하려면 docker inspect 명령을 사용

아래 스샷은 리소스 제한이 없는 Container 의 Memory 와 Cpu 상태를 보여줌 (리소스 제한이 없어서 0으로 나타남)

Container 에 memory 제한을 주려면 다음과 같은 옵션을 사용

"Memory" 에 숫자가 나타나며, 숫자가 1gb 크기를 보여주는 것을 확인 가능

"MemorySwap" 을 따로 지정해주지 않으면 "Memory" 의 2배 크기만큼 잡힌다고 함

"MemorySwap" 을 따로 지정하려면 아래 옵션 추가

이 외에 --memory-swappiness, --kernel-memory, --oom-kill-disable 등의 옵션이 존재함

Container 에 cpu 제한을 주려면 다음과 같은 옵션을 사용

이 외에 --cpu-period, --cpu-quata, --cpu-shares  등의 옵션이 존재함

리소스 제한 설정을 하지 않은 상태로 이미 올라간 Container 에

리소스를 제한하고 싶다면 update 명령어 사용

참고로 부하 테스트를 진행할때 사용할 수 있는 stress 라는 명령어가 존재함

cpu 부하, memory 부하를 임의로 일으킬 수 있음

Docker 에 대한 내용이 너무 길어져서

Docker Compose 에 대한 내용은 다른 포스트로 올림

https://eyeballs.tistory.com/642

Airflow 테스트를 위해

Docker 로 Airflow 설치하는 방법을 설명함

Centos 7 을 사용하였고, Airflow 는 Sequential Executor 를 사용함.

Centos 7 Docker Image : https://eyeballs.tistory.com/543

위 Docker Image 를 만들어준 다음, 아래 명령어로 container 하나 띄움

아래 명령어로 docker container 들어가서 필요한 것들 설치 진행

아래 명령어로 sqlite 최신 버전 설치 [참고]

( sqlite 다운로드 페이지 :https://www.sqlite.org/download.html )

만약 sqlite3 --version 이 3.7.17

위처럼 sqlite3 를 최신 버전으로 업그레이드 하는 경우는

pip 로 airflow 설치시 3.8.3 이상의 sqlite 버전을 필요로 하기 때문.

can't find new sqlite version? (SQLite 3.8.3 or later is required (found 3.7.17))

만약 sqlite3 버전이 3.8.3 이상이라면 위와 같이 최신 버전으로 업그레이드 하지 않아도 됨

airflow 라는 사용자를 만들어서 sudo 권한 부여

아래 명령어로 /etc/sudoers 편집 진행

아래 주석 처리되어있는 부분의 주석을 해제

아래 명령어로 사용자 airflow 로 접근

지금부터 나오는 명령어들은 모두 airflow 계정으로 진행

아래 명령어로 pip3 업그레이드 진행 및 setuptool 설치

아래 명령어로 airflow home 을 만듦

아래 명령어로 sqlite3 가 최신 버전(>3.7.17)으로 업그레이드 되었는지 확인

아래 명령어로 airflow 설치

나는 2.2.2를 설치하고 싶어서 직접 2.2.2를 넣었으니, 원하는 버전을 넣으면 됨.

아래 명령어로 apache airflow 의 admin 유저 생성

아래 명령어로 apache airflow 를 Sequential Executor 모드로 실행

잠시 후 웹브라우저에서 localhost:8080 에 접근하여 WebUI 가 뜨는지 확인

id, pw 는 각각 airflow, airflow 로 접근 가능

dags 위치는 airflow.cfg 에서 찾아볼 수 있음 ($AIRFLOW_HOME/dags)

참고

https://airflow.apache.org/docs/apache-airflow/2.2.2/start/local.html

https://www.webdesignsun.com/insights/upgrading-sqlite-on-centos/

https://musclebear.tistory.com/131

https://sun2day.tistory.com/216

"Data Pipellines with Apache Airflow

Apache Airflow 기반의 데이터 파이프라인

에어플로 중심의 워크플로 구축에서 커스텀 컴포넌트 개발 및 배포, 관리까지"

책 내용 공부한 후, 나중에 다시 참고할 내용 필기해 둠

책은 Airflow 2.0.0 버전을 기준으로 쓰여졌지만,

나는 Airflow 2.2.2 버전에서 dag 구현 및 테스트 진행함

기본 DummyOperator, BashOperator, PythonOperator 샘플 코드

from datetime import datetime, timedelta

from airflow import DAG
from airflow.operators.bash import BashOperator
from airflow.operators.python import PythonOperator
from airflow.operators.dummy import DummyOperator

def _func():
    print('hello python!')

with DAG(
    dag_id='eyeballs_dag',
    schedule_interval='0 0 * * *',
    start_date=datetime(2023, 6, 25),
    catchup=False,
    dagrun_timeout=timedelta(minutes=60),
    tags=['eyeballs'],
    params={"example_key": "example_value"},
) as dag:

    start = DummyOperator(
        task_id='dummy_operator',
    )

    bash_operator = BashOperator(
        task_id='bash_operator',
        bash_command='echo hello bash!',
    )

    python_operator = PythonOperator(
        task_id='python_operator',
        python_callable=_func,
    )

    start >> bash_operator >> python_operator

dag 에 schedule_interval 은 cron 으로 넣을 수 있지만,

@daily 등으로 넣을 수 있음 [참고]

with DAG(
    dag_id='eyeballs_dag',
    schedule_interval='@daily',
    start_date=datetime(2023, 6, 25),
) as dag:

종료 날짜를 지정하면, 종료 날짜까지만 dag 스케줄링을 진행

예를 들어 아래와 같이 1월 5일까지만 지정하면

execution_date 가 1월 5일 값을 갖는 날까지(즉 현실 시간으로 1월 6일) 동작

with DAG(
    dag_id='eyeballs_dag',
    schedule_interval='@daily',
    start_date=datetime(2023, 1, 1),
    end_date=datetime(2023, 1, 5)
) as dag:

며칠 몇 시에 동작하도록 스케줄링하는 게 아니라

3시간마다 혹은 10분마다 스케줄링하도록 만들려면

아래 코드와 같이 dt.timedelta 를 사용하면 됨

3일마다 스케줄링 : dt.timedelta(days=3)

2시간마다 스케줄링 : dt.timedelta(hours=2)

with DAG(
    dag_id='eyeballs_dag',
    schedule_interval='dt.timedelta(days=3)',
    start_date=datetime(2023, 6, 25),
) as dag:

jinja template 진자 템플릿을 이용하여

airflow 가 기본적으로 갖고 있는 변수 중 하나인

execution_date 을 bash operator 에서 사용하는 샘플 코드

아래 dag 를 실행하는 실제 날짜는 10월 1일인 상황

코드 :

    b1 = BashOperator(
        task_id='b1',
        bash_command='''
            echo execution_date : {{execution_date}} ;
            echo ds : {{ds}} ;
            echo ds_nodash : {{ds_nodash}} ;
            echo execution_date with strftime : {{execution_date.strftime("%Y_%m_%d")}}
        '''
    )

결과 :

[2023-10-01, 13:43:54 UTC] {subprocess.py:89} INFO - execution_date : 2023-09-30T00:00:00+00:00
[2023-10-01, 13:43:54 UTC] {subprocess.py:89} INFO - ds : 2023-09-30
[2023-10-01, 13:43:54 UTC] {subprocess.py:89} INFO - ds_nodash : 20230930
[2023-10-01, 13:43:54 UTC] {subprocess.py:89} INFO - execution_date with strftime : 2023_09_30

코드 :

    b2 = BashOperator(
        task_id='b2',
        bash_command='''
            echo next_execution_date : {{next_execution_date}} ;
            echo next_ds : {{next_ds}} ;
            echo next_ds_nodash : {{next_ds_nodash}} ;
        '''
    )

결과 :

[2023-10-01, 13:43:58 UTC] {subprocess.py:89} INFO - next_execution_date : 2023-10-01T00:00:00+00:00
[2023-10-01, 13:43:58 UTC] {subprocess.py:89} INFO - next_ds : 2023-10-01
[2023-10-01, 13:43:58 UTC] {subprocess.py:89} INFO - next_ds_nodash : 20231001

코드 :

    b3 = BashOperator(
        task_id='b3',
        bash_command='''
            echo prev_execution_date : {{prev_execution_date}} ;
            echo prev_ds : {{prev_ds}} ;
            echo prev_ds_nodash : {{prev_ds_nodash}}
        '''
    )

결과 :

[2023-10-01, 13:44:01 UTC] {subprocess.py:89} INFO - prev_execution_date : 2023-09-29 00:00:00+00:00
[2023-10-01, 13:44:01 UTC] {subprocess.py:89} INFO - prev_ds : 2023-09-29
[2023-10-01, 13:44:01 UTC] {subprocess.py:89} INFO - prev_ds_nodash : 20230929

airflow 가 갖는 기본 변수 : https://airflow.apache.org/docs/apache-airflow/2.2.2/templates-ref.html

airflow 가 갖는 기본 변수는 다음과 같이 코드에서 직접 확인 가능

코드 :

def _func(**kwargs):
    for arg in kwargs:
        print(arg, ':', kwargs[arg])

...


p1 = PythonOperator(
    task_id='p1',
    python_callable=_func,
    dag=dag
)

결과 :

[2023-10-01, 14:28:52 UTC] {logging_mixin.py:109} INFO - conf : <airflow.configuration.AirflowConfigParser object at 0x7f9bc30a6eb8>
[2023-10-01, 14:28:52 UTC] {logging_mixin.py:109} INFO - dag : <DAG: execution_date>
[2023-10-01, 14:28:52 UTC] {logging_mixin.py:109} INFO - dag_run : <DagRun execution_date @ 2023-10-01 13:23:29.703212+00:00: manual__2023-10-01T13:23:29.703212+00:00, externally triggered: True>
[2023-10-01, 14:28:52 UTC] {logging_mixin.py:109} INFO - data_interval_end : 2023-10-01T00:00:00+00:00
[2023-10-01, 14:28:52 UTC] {logging_mixin.py:109} INFO - data_interval_start : 2023-09-30T00:00:00+00:00
[2023-10-01, 14:28:52 UTC] {logging_mixin.py:109} INFO - ds : 2023-10-01
[2023-10-01, 14:28:52 UTC] {logging_mixin.py:109} INFO - ds_nodash : 20231001
[2023-10-01, 14:28:52 UTC] {logging_mixin.py:109} WARNING - /home/airflow/.local/lib/python3.6/site-packages/airflow/models/taskinstance.py:1941 DeprecationWarning: Accessing 'execution_date' from the template is deprecated and will be removed in a future version. Please use 'logical_date' or 'data_interval_start' instead.
[2023-10-01, 14:28:52 UTC] {logging_mixin.py:109} INFO - execution_date : 2023-10-01T13:23:29.703212+00:00
[2023-10-01, 14:28:52 UTC] {logging_mixin.py:109} INFO - inlets : []
[2023-10-01, 14:28:52 UTC] {logging_mixin.py:109} INFO - logical_date : 2023-10-01T13:23:29.703212+00:00
[2023-10-01, 14:28:52 UTC] {logging_mixin.py:109} INFO - macros : <module 'airflow.macros' from '/home/airflow/.local/lib/python3.6/site-packages/airflow/macros/__init__.py'>
[2023-10-01, 14:28:52 UTC] {logging_mixin.py:109} WARNING - /home/airflow/.local/lib/python3.6/site-packages/airflow/models/taskinstance.py:1941 DeprecationWarning: Accessing 'next_ds' from the template is deprecated and will be removed in a future version. Please use 'data_interval_end | ds' instead.
[2023-10-01, 14:28:52 UTC] {logging_mixin.py:109} INFO - next_ds : 2023-10-01
[2023-10-01, 14:28:52 UTC] {logging_mixin.py:109} WARNING - /home/airflow/.local/lib/python3.6/site-packages/airflow/models/taskinstance.py:1941 DeprecationWarning: Accessing 'next_ds_nodash' from the template is deprecated and will be removed in a future version. Please use 'data_interval_end | ds_nodash' instead.
[2023-10-01, 14:28:52 UTC] {logging_mixin.py:109} INFO - next_ds_nodash : 20231001
[2023-10-01, 14:28:52 UTC] {logging_mixin.py:109} WARNING - /home/airflow/.local/lib/python3.6/site-packages/airflow/models/taskinstance.py:1941 DeprecationWarning: Accessing 'next_execution_date' from the template is deprecated and will be removed in a future version. Please use 'data_interval_end' instead.
[2023-10-01, 14:28:52 UTC] {logging_mixin.py:109} INFO - next_execution_date : 2023-10-01T13:23:29.703212+00:00
[2023-10-01, 14:28:52 UTC] {logging_mixin.py:109} INFO - outlets : []
[2023-10-01, 14:28:52 UTC] {logging_mixin.py:109} INFO - params : {'example_key': 'example_value'}
[2023-10-01, 14:28:52 UTC] {logging_mixin.py:109} INFO - prev_data_interval_start_success : 2023-09-30T00:00:00+00:00
[2023-10-01, 14:28:52 UTC] {logging_mixin.py:109} INFO - prev_data_interval_end_success : 2023-10-01T00:00:00+00:00
[2023-10-01, 14:28:52 UTC] {logging_mixin.py:109} WARNING - /home/airflow/.local/lib/python3.6/site-packages/airflow/models/taskinstance.py:1941 DeprecationWarning: Accessing 'prev_ds' from the template is deprecated and will be removed in a future version.
[2023-10-01, 14:28:52 UTC] {logging_mixin.py:109} INFO - prev_ds : 2023-10-01
[2023-10-01, 14:28:52 UTC] {logging_mixin.py:109} WARNING - /home/airflow/.local/lib/python3.6/site-packages/airflow/models/taskinstance.py:1941 DeprecationWarning: Accessing 'prev_ds_nodash' from the template is deprecated and will be removed in a future version.
[2023-10-01, 14:28:52 UTC] {logging_mixin.py:109} INFO - prev_ds_nodash : 20231001
[2023-10-01, 14:28:52 UTC] {logging_mixin.py:109} WARNING - /home/airflow/.local/lib/python3.6/site-packages/airflow/models/taskinstance.py:1941 DeprecationWarning: Accessing 'prev_execution_date' from the template is deprecated and will be removed in a future version.
[2023-10-01, 14:28:52 UTC] {logging_mixin.py:109} INFO - prev_execution_date : 2023-10-01T13:23:29.703212+00:00
[2023-10-01, 14:28:52 UTC] {logging_mixin.py:109} WARNING - /home/airflow/.local/lib/python3.6/site-packages/airflow/models/taskinstance.py:1941 DeprecationWarning: Accessing 'prev_execution_date_success' from the template is deprecated and will be removed in a future version. Please use 'prev_data_interval_start_success' instead.
[2023-10-01, 14:28:52 UTC] {logging_mixin.py:109} INFO - prev_execution_date_success : None
[2023-10-01, 14:28:52 UTC] {logging_mixin.py:109} INFO - prev_start_date_success : 2023-10-01T13:43:50.677389+00:00
[2023-10-01, 14:28:52 UTC] {logging_mixin.py:109} INFO - run_id : manual__2023-10-01T13:23:29.703212+00:00
[2023-10-01, 14:28:52 UTC] {logging_mixin.py:109} INFO - task : <Task(PythonOperator): p1>
[2023-10-01, 14:28:52 UTC] {logging_mixin.py:109} INFO - task_instance : <TaskInstance: execution_date.p1 manual__2023-10-01T13:23:29.703212+00:00 [running]>
[2023-10-01, 14:28:52 UTC] {logging_mixin.py:109} INFO - task_instance_key_str : execution_date__p1__20231001
[2023-10-01, 14:28:52 UTC] {logging_mixin.py:109} INFO - test_mode : False
[2023-10-01, 14:28:52 UTC] {logging_mixin.py:109} INFO - ti : <TaskInstance: execution_date.p1 manual__2023-10-01T13:23:29.703212+00:00 [running]>
[2023-10-01, 14:28:52 UTC] {logging_mixin.py:109} WARNING - /home/airflow/.local/lib/python3.6/site-packages/airflow/models/taskinstance.py:1941 DeprecationWarning: Accessing 'tomorrow_ds' from the template is deprecated and will be removed in a future version.
[2023-10-01, 14:28:52 UTC] {logging_mixin.py:109} INFO - tomorrow_ds : 2023-10-02
[2023-10-01, 14:28:52 UTC] {logging_mixin.py:109} WARNING - /home/airflow/.local/lib/python3.6/site-packages/airflow/models/taskinstance.py:1941 DeprecationWarning: Accessing 'tomorrow_ds_nodash' from the template is deprecated and will be removed in a future version.
[2023-10-01, 14:28:52 UTC] {logging_mixin.py:109} INFO - tomorrow_ds_nodash : 20231002
[2023-10-01, 14:28:52 UTC] {logging_mixin.py:109} INFO - ts : 2023-10-01T13:23:29.703212+00:00
[2023-10-01, 14:28:52 UTC] {logging_mixin.py:109} INFO - ts_nodash : 20231001T132329
[2023-10-01, 14:28:52 UTC] {logging_mixin.py:109} INFO - ts_nodash_with_tz : 20231001T132329.703212+0000
[2023-10-01, 14:28:52 UTC] {logging_mixin.py:109} INFO - var : {'json': None, 'value': None}
[2023-10-01, 14:28:52 UTC] {logging_mixin.py:109} INFO - conn : <airflow.models.taskinstance.TaskInstance.get_template_context.<locals>.ConnectionAccessor object at 0x7f9bc06c3748>
[2023-10-01, 14:28:52 UTC] {logging_mixin.py:109} WARNING - /home/airflow/.local/lib/python3.6/site-packages/airflow/models/taskinstance.py:1941 DeprecationWarning: Accessing 'yesterday_ds' from the template is deprecated and will be removed in a future version.
[2023-10-01, 14:28:52 UTC] {logging_mixin.py:109} INFO - yesterday_ds : 2023-09-30
[2023-10-01, 14:28:52 UTC] {logging_mixin.py:109} WARNING - /home/airflow/.local/lib/python3.6/site-packages/airflow/models/taskinstance.py:1941 DeprecationWarning: Accessing 'yesterday_ds_nodash' from the template is deprecated and will be removed in a future version.
[2023-10-01, 14:28:52 UTC] {logging_mixin.py:109} INFO - yesterday_ds_nodash : 20230930
[2023-10-01, 14:28:52 UTC] {logging_mixin.py:109} INFO - templates_dict : None

airflow 2.2 버전부터 헷갈리는 execution_date 대신 data_interval_start 를 사용하도록 업데이트 됨

Airflow dag 를 작성할 때 가장 중요하게 고려해야 하는 두 가지는 원자성과 멱등성

원자성 : Airflow 태스크 수행이 성공하려면 완벽하게 성공해야 하고, 실패하려면 완벽하게 실패해야 함

멱등성 : 동일한 입력으로 동일한 태스크를 여러 번 호출해도 결과가 동일해야 함

zero-padding : 빈 앞자리를 0으로 채움

예를 들어 1월이면 01, 8일이면 08, 4시면 04시

execution_date 에서 추출한 월, 일, 시간에 zero-padding 을 넣는 방법을 다음과 같음

코드 :

    b1 = BashOperator(
        task_id='b1',
        bash_command='''
            echo execution_date : {{execution_date}} ;
            echo execution_date.year: {{execution_date.year}} ;
            echo execution_date.month: {{execution_date.month}} ;
            echo month with zero-padding : {{ '{:02}'.format(execution_date.month) }} ;
            echo execution_date.day: {{execution_date.day}} ;
            echo day with zero-padding : {{ '{:02}'.format(execution_date.day) }} ;
            echo execution_date.hour: {{execution_date.hour}} ;
            echo hour with zero-padding : {{ '{:02}'.format(execution_date.hour) }} ;
        '''
    )

결과 :

execution_date : 2023-10-01T07:23:29.703212+00:00
execution_date.year: 2023
execution_date.month: 10
month with zero-padding : 10
execution_date.day: 1
day with zero-padding : 01
execution_date.hour: 7
hour with zero-padding : 07

PythonOperator 가 실행하는 함수에서 execution_date 를 사용하는 여러가지 방법

def _func1(execution_date):
    print('execution_date : ', execution_date)
    year, month, day, hour, *_ = execution_date.timetuple()
    print('year : ', year)
    print('month : ', month)
    print('day : ', day)
    print('hour : ', hour)

def _func2(**context):
    print('execution_date : ', context['execution_date'])

def _func3(execution_date, **context):
    print('execution_date : ', execution_date)
    print('execution_date' in context) #False

...

    p1 = PythonOperator(
        task_id='p1',
        python_callable=_func1,
    )

    p2 = PythonOperator(
        task_id='p2',
        python_callable=_func2,
    )

    p3 = PythonOperator(
        task_id='p3',
        python_callable=_func3,
    )

p3 처럼 execution_date 를 파라미터에 직접 넣었다면

**context 에 execution_date 값은 불포함

PythonOperator 가 실행하는 함수에 변수 넣는 방법

def _func1(greeting, name, date):
    print(f"{greeting}! I'm {name}. Today is {date}")

def _func2(param1, param2, param3, **context):
    print(f"{param1}! I'm {param2}. Today is {param3}")

...

    p1 = PythonOperator(
        task_id='p1',
        python_callable=_func1,
        op_kwargs={
            "greeting":"hello",
            "name":"eyeballs",
            "date":"{{execution_date}}"
        }
    )

    p2 = PythonOperator(
        task_id='p2',
        python_callable=_func2,
        op_args=["hello", "eyeballs", "{{execution_date}}"]
    )

func1 결과 :

[2023-10-02, 00:31:57 UTC] {logging_mixin.py:109} INFO - hello! I'm eyeballs. Today is 2023-10-01T00:00:00+00:00

func2 결과 :

[2023-10-02, 00:32:00 UTC] {logging_mixin.py:109} INFO - hello! I'm eyeballs. Today is 2023-10-01T00:00:00+00:00

아래와 같이 Rendered 를 누르면 Python Operator 에서 함수로 넘겨준 파라미터들을 볼 수 있음

Airflow Task 간 데이터를 교환/전송할 때 xcom 을 사용할 수 있음

하나의 Task 가 xcom(피클링)을 통해 데이터를 메타스토어에 저장하고,

다른 Task 에서 xcom 이 저장한 데이터를 메타스토어로부터 읽는 것으로 데이터를 교환함

피클 Picke 이란, Python 객체를 디스크에 저장할 수 있게 하는 직렬화 프로토콜

피클링된 객체는 메타스토어의 블록 blob 에 저장되기 때문에

일반적으로 문자열 몇 개와 같은 작은 데이터 전송에만 적용 가능

Task 간 좀 더 큰 데이터를 교환 할 때는 xcom 이 아닌 외부 데이터베이스를 사용해야 함

xcom 샘플 코드

def _push_xcom(**context):
    context["task_instance"].xcom_push(key="name", value="eyeballs")

def _pull_xcom(**context):
    name=context["task_instance"].xcom_pull(task_ids="push_op", key="name")
    print(f"Hello, {name}!")

...

    push_op = PythonOperator(
        task_id='push_op',
        python_callable=_push_xcom,
    )

    pull_op = PythonOperator(
        task_id='pull_op',
        python_callable=_pull_xcom,
    )

    push_op >> pull_op

위 코드에서 사용된 xcom_pull 은 현재 DAG 실행을 통해 게시된 값만 가져옴

다른 날짜에서 실행된 task 가 xcom_push 로 넣은 값을 (키값이 동일해도) 가져오지 않는다는 말임

의존성 fan-out, fan-in

a >> [b,c] >> d

트리거 규칙 Trigger Rule 이란, Task 가 트리거되는 기준을 의미함

즉, Task 가 언제 어떤 조건이 되었을 때 실행 될 것인지를 말 함

airflow Trigger Rule 문서 :

https://airflow.apache.org/docs/apache-airflow/2.2.2/concepts/dags.html?highlight=all_failed#trigger-rules 

• all_success (default): All upstream tasks have succeeded
  모든 상위 태스크가 성공하면 트리거됨
• all_failed: All upstream tasks are in a failed or upstream_failed state
  모든 상위 태스크가 실패하면 트리거됨
  태스크 그룹에서 하나 이상 실패가 예상되는 상황에서 오류 처리 코드를 트리거하는 용도로 사용
• all_done: All upstream tasks are done with their execution
  성공/실패 상관 없이 모든 상위 태스크가 실행을 마치면 트리거됨
  모든 상위 태스크 실행 완료 후 청소 코드 등을 실행하는 용도로 사용
• one_failed: At least one upstream task has failed (does not wait for all upstream tasks to be done)
  하나 이상의 상위 태스크가 실패하자마자 트리거됨. 다른 상위 태스크의 실행 완료는 기다리지 않음
  알림 또는 롤백 같은 일부 오류 처리 코드를 빠르게 실행하기 위한 용도로 사용
• one_success: At least one upstream task has succeeded (does not wait for all upstream tasks to be done)
  하나 이상의 상위 태스크가 성공하자마자 트리거됨. 다른 상위 태스크의 실행 완료는 기다리지 않음
  하나의 결과를 사용할 수 있게 되는 즉시 다운스트림 연산/알람을 빠르게 실행하기 위한 용도로 사용
• none_failed: All upstream tasks have not failed or upstream_failed - that is, all upstream tasks have succeeded or been skipped
  실패한 상위 태스크는 없지만, 태스크가 성공 또는 건너 뛴 경우 트리거됨
  DAG 에 조건부 브랜치 구조가 있는 경우 사용
• none_failed_min_one_success: All upstream tasks have not failed or upstream_failed, and at least one upstream task has succeeded.
• none_skipped: No upstream task is in a skipped state - that is, all upstream tasks are in a success, failed, or upstream_failed state
  건너뛴 상위 태스크는 없지만, 태스크가 성공 또는 실패한 경우 트리거됨
  모든 업스트림 태스크가 실행된 경우, 해당 결과를 무시하고 실행하기 위한 용도로 사용
• always: No dependencies at all, run this task at any time
  상위 태스크 상태와 상관없이 트리거됨
  테스트 용도로 사용

센서는 특정 조건이 true 인지 지속적으로 확인하고 true 라면 성공

만약 특정 조건에 false 가 뜨면, 상태가 true 가 될 때 까지 혹은 지정된 timeout 이 날 때까지 계속 확인함

PythonSensor 의 경우, PythonSensor가 사용하는 함수의 return 값이 true 인지 false 인지를 계속 확인함

PythonSensor 샘플 코드

from datetime import datetime, timedelta

from airflow import DAG
from airflow.operators.python import PythonOperator
from airflow.sensors.python import PythonSensor
from airflow.operators.dummy import DummyOperator

def _sensor():
    if datetime.now().second == 30 : return True
    else : return False

with DAG(
    dag_id='sensor',
    schedule_interval='0 0 * * *',
    start_date=datetime(2023, 6, 25),
    catchup=False,
) as dag:

    start_op = DummyOperator(
        task_id='start_op',
    )

    sensor_op = PythonSensor(
        task_id='sensor_op',
        python_callable=_sensor,
    )

    stop_op = DummyOperator(
        task_id='stop_op',
    )

    start_op >> sensor_op >> stop_op

sensor 가 무한정 기다리는 상황이 닥치게 되면

그 다음 스케줄링 때 다시 sensor 가 실행되고

그 다음 스케줄링 떄 다시 sensor 가 실행되고...

이렇게 sensor 가 끝나지 않고 계속 누적되어 동작하여

DAG 내의 최대 task 수 (default 16) 개에 도달하게 되면

새로운 task 가 실행 자체를 못 하는 경우가 발생함

따라서 DAG 내 최대 task 수를 조절하거나

sensor 의 timeout 을 하루 등으로 줄이거나

poke 모드 대신 reschedule 모드를 사용하여 포크 동작을 실행할 때만 task 슬롯을 차지하고

대기 시간 동안은 task 슬롯을 차지하지 않도록 만들어야 함

Dag 에서 다른 Dag 를 Trigger 할 수 있음

TriggerDagRunOperator 샘플 코드

from datetime import datetime, timedelta

from airflow import DAG
from airflow.operators.dummy import DummyOperator
from airflow.operators.trigger_dagrun import TriggerDagRunOperator

with DAG(
    dag_id='dag1',
    schedule_interval='0 0 * * *',
    start_date=datetime(2023, 6, 25),
) as dag1:
    start_op = DummyOperator(
        task_id='start_op',
    )
    trigger_dag = TriggerDagRunOperator(
        task_id='trigger_dag',
        trigger_dag_id='dag2'
    )
    start_op >> trigger_dag 

with DAG(
    dag_id='dag2',
    schedule_interval=None,
    start_date=datetime(2023,6,25),
) as dag2:
    start_op = DummyOperator(
        task_id='start_op',
    )
    start_op

결과 : dag2 의 Schedule 은 None 임에도 불구하고, dag1 에 의해 호출되어 실행됨

Dag 간 fan-in 의존성 맞추는 방법은 시스템 상에서 제공되지 않음

예를 들어 dag1과 dag2가 모두 성공해야 dag3 이 트리거되는 방법은 없다는 말임

대신 xcom 등으로 dag1, dag2 의 성공 여부를 저장하고 dag3 에서 센서 등으로 결과값을 확인한다거나

ExternalTaskSensor 를 사용하여 dag1, dag2 의 상태를 꾸준히 확인하는 방법을 사용 가능함

Airflow ExternalTaskSensor Documentation : https://airflow.apache.org/docs/apache-airflow/2.2.2/howto/operator/external_task_sensor.html

ExternalTaskSensor 를 사용하는 코드 예제 (테스트 진행되지 않았으니, 반드시 테스트 진행 후 사용해야 함)

from datetime import datetime, timedelta

from airflow import DAG
from airflow.operators.dummy import DummyOperator
from airflow.sensors.external_task import ExternalTaskSensor

with DAG(
    dag_id='dag1',
    schedule_interval='0 0 * * *',
    start_date=datetime(2023, 6, 25),
    catchup=False,
) as dag1:
    dag1_op = DummyOperator(
        task_id='dag1_op',
    )

with DAG(
    dag_id='dag2',
    schedule_interval="0 0 * * *",
    start_date=datetime(2023,6,25),
    catchup=False,
) as dag2:
    dag2_op = DummyOperator(
        task_id='dag2_op',
    )

with DAG(
    dag_id='dag3',
    schedule_interval=None,
    start_date=datetime(2023,6,25),
    catchup=False,
) as dag3:
    dag1_sensor = ExternalTaskSensor(
        task_id = "dag1_sensor",
        external_dag_id = "dag1",
        external_task_id = "dag1_op"
    )
    dag2_sensor = ExternalTaskSensor(
        task_id = "dag2_sensor",
        external_dag_id = "dag2",
        external_task_id = "dag2_op"
    )
    start_op = DummyOperator(
        task_id='start_op',
    )
    [dag1_sensor, dag2_sensor] >> start_op

Task, DAG의 실행이 평소보다 오래 걸리는 상황을 파악하기 위해

SLA(Service-Level Agreement) 을 추가 가능

SLA 로 지정해 둔 제한 시간보다 실행 시간이 오래 걸린다면 경고를 주거나 python 함수 등을 실행하게 만듦

SLA는 Dag 및 Task 에 각각 설정 가능함

Task 의 시작 또는 종료 시간이 DAG 시작 시간고 비교하여 SLA 에 지정한 제한 시간을 넘겼는지 확인 할 것임

SLA 를 2초로 설정한 코드 예제 (테스트 진행되지 않았으니, 반드시 테스트 진행 후 사용해야 함)

from datetime import datetime, timedelta

from airflow import DAG
from airflow.operators.python import PythonOperator
from airflow.operators.dummy import DummyOperator

default_args={
    "sla":timedelta(seconds=2),
}

def sla_miss_callback(context):
    print("TOO LONG!")

def _func():
    sum = 0
    for i in range(99999999):
        sum = sum + i
    print(sum)

with DAG(
    dag_id='test',
    schedule_interval='0 0 * * *',
    start_date=datetime(2023, 6, 25),
    catchup=False,
    sla_miss_callback=sla_miss_callback,
    default_args=default_args,
) as dag:

    start = DummyOperator(
        task_id='dummy_operator',
    )
    
    python = PythonOperator(
        task_id='python',
        python_callable=_func
    )

    start >> python

다음과 같이 Operator 에 적용할 수 있다고 함

    python = PythonOperator(
        task_id='python',
        python_callable=_func,
        sla=timedelta(seconds=2)
    )

DAG 의 시작 시간과 Task 의 종료 시간을 계속 비교한다고 함

주의 할 점은 비교하는 시작 시간이 Task 의 시작 시간이 아니라 DAG 의 시작 시간이라는 것

(개별 Task 가 아닌 DAG 시작 시간을 기준으로 SLA 가 정의되었기 때문)

sla 뿐만 아니라, Task 가 성공했을 때 혹은 Task 가 실패했을 때 callback 을 호출하도록 만들 수 있음

Callback 사용 샘플 코드

from datetime import datetime, timedelta
from airflow import DAG
from airflow.operators.dummy import DummyOperator


def task_failure_alert(context):
    print(f"Task has failed, task_instance_key_str: {context['task_instance_key_str']}")


def dag_success_alert(context):
    print(f"DAG has succeeded, run_id: {context['run_id']}")


with DAG(
    dag_id="example_callback",
    schedule_interval=None,
    start_date=datetime(2021, 1, 1),
    dagrun_timeout=timedelta(minutes=60),
    catchup=False,
    on_success_callback=None,
    on_failure_callback=task_failure_alert,
    tags=["example"],
) as dag:

    task1 = DummyOperator(task_id="task1")
    task2 = DummyOperator(task_id="task2")
    task3 = DummyOperator(task_id="task3", on_success_callback=dag_success_alert)
    task1 >> task2 >> task3

Task 들을 그룹으로 묶어서

눈으로 보기에 좀 더 정렬되어 보이고 이해하기 쉽도록 만들 수 있음

Task Group 사용 샘플 코드

from datetime import datetime

from airflow.models.dag import DAG
from airflow.operators.bash import BashOperator
from airflow.operators.dummy import DummyOperator
from airflow.utils.task_group import TaskGroup

# [START howto_task_group]
with DAG(
    dag_id="example_task_group", start_date=datetime(2021, 1, 1), catchup=False, tags=["example"]
) as dag:
    start = DummyOperator(task_id="start")

    # [START howto_task_group_section_1]
    with TaskGroup("section_1", tooltip="Tasks for section_1") as section_1:
        task_1 = DummyOperator(task_id="task_1")
        task_2 = BashOperator(task_id="task_2", bash_command='echo 1')
        task_3 = DummyOperator(task_id="task_3")

        task_1 >> [task_2, task_3]
    # [END howto_task_group_section_1]

    # [START howto_task_group_section_2]
    with TaskGroup("section_2", tooltip="Tasks for section_2") as section_2:
        task_1 = DummyOperator(task_id="task_1")

        # [START howto_task_group_inner_section_2]
        with TaskGroup("inner_section_2", tooltip="Tasks for inner_section2") as inner_section_2:
            task_2 = BashOperator(task_id="task_2", bash_command='echo 1')
            task_3 = DummyOperator(task_id="task_3")
            task_4 = DummyOperator(task_id="task_4")

            [task_2, task_3] >> task_4
        # [END howto_task_group_inner_section_2]

    # [END howto_task_group_section_2]

    end = DummyOperator(task_id='end')

    start >> section_1 >> section_2 >> end

Airflow WebUI Graph 에서 보이는 화면

차례대로 Group 을 클릭하지 않았을 때, Group 을 클릭했을 때

전치사 활용한 표현

Im speaking from my own experience.

Difficult to draw any definite conclusion from this evidence aloneFrom the look of things, this will take much longer than we thought.I can figure out one conclusion from this truth.From : (~에) 기초하여, (~을) 토대로draw : (결론을) 내리다

From the look of things : 상황을 보니..

Don't drink milk from the milkbox.

She drank water from the bottle.

My cat likes drinking from the faucet!

From : (~에) 대로 (마시는)

You can build yourself a nice PC for 300 dollars.

You can buy a nice house on the island for 20 billion dollars

For : (~의 돈을) 들이면

I'm all for the idea! 그 아이디어에 찬성!

Are you for or against the proposal??

For : (~에) 찬성하는

I'm in a yoga class.

She was in the univ.

My child is in middle school.

When I was in colleage, I waited tables at a family restaurant.

In : (~에) 재학중인, 다니고 있는

wait tables : 종업원으로 일하다

Usually I'm in bed by 9

She is in bed now.

In bed : 잠자리에 든

It's not in our interest to use your service.

This law is not in the interest of consumers.

In the interest : (~의) 이익에 부합하는

I'm in love.

She is in a coma.

Are you in pain?

The economy is in a deep slump...

In : (~에) 빠져있는

She looked at herself in the mirror.

He caught a glimpse of a beautiful woman in the mirror.

In the mirror : 거울에 비친

catch a glimpse of : ~을 흘끗 보다, 언뜻 보다

glimpse : (완전히는 못 보고) 잠깐[언뜻] 봄, 일별

I work in the IT field.

He is in the Internet business!

My family has been in the food business for generations.

In : ~의 사업을 하고 있는, ~의 분야에 종사하는

for generations : 여러 세대 동안