1. 문제 상황 (에러 코드)

Custom OAuth is not configured for this datasource. Learn more. (errorCode=170001)

• 결론적으로, BigQuery 커넥터에 구글 OAuth client ID로 태블로 서버에 구성이 필요함 (링크)

2. 해결 방법

• Step1. Google OAuth Client ID 및 Redirect URL 생성
• Step2. 태블로 서버에 생성한 Oauth Client ID와 Redirect URL 등록

[Step1]

• 구글 OAuth 설정을 위한 프로젝트 생성 ( [Set up OAuth for Google]

  • APIs & Services > OAuth consent screen 클릭
    • 유저 타입을 설정하고 이후 작성 항목들에 대하여 입력

• APIs & Services > Credentials (사용자 인증 정보) 등록

  • Create Credentials > OAuth client ID

• Application type에 태블로의 어플리케이션 타입을 입력 → Web application

  • 권한이 승인된 자바스크립 원본 → https 나 http로 시작하는 태블로 웹 페이지 URL을 추가함

  • 승인된 리디렉션 원본 → https 나 http로 시작하는 웹페이지 URL + /auth/add_oauth_token 을 추가

    https://your_server_url/auth/add_oauth_token

[Step2]

• Redirect URL을 이용하여 로컬PC에서 태블로에 접근하여 Client ID와 비밀번호를 입력해서 등록

• Register the OAuth client ID and client secret!

  • 관리자 권한 계정으로 태블로 웹페이지 서버로 들어가서 , 위에 설정한 정보들을 입력하여 등록

       ![](https://velog.velcdn.com/images/sophi_e/post/b6aa64c8-288d-48f6-92c9-f3c9072b88ac/image.png)

절차적 언어란?

• one query에서 다중 statement를 multi statements query로 실행할 수 있음
• multiple statements 쿼리를 사용하여 다음을 수행
  • 순차적으로 multi statements를 수행함
  • 테이블들을 생성/삭제와 같은 관리 작업을 자동화함
  • IF와 WHILE과 같은 프로그래밍 구조를 사용하여 복잡한 로직을 실행함

1. DECLARE

DECLARE variable_name[, ...] [variable_type] [DEFAULT expression];

• variable_name은 타당한 구분자가 되어야만 하며, variable type은 GoogleSQL type 중 하나 (링크)

• DECLARE 변수명을 기재하는 구문

• DEFAULT

  • 지정되면, 변수는 그 expression의 값으로 초기화 되지만, 만약 default값이 지정되지 않으면 변수는 null값으로 초기화됨
  • variable_type이 생략된다면, DEFAULT가 반드시 반드시 상세화되어야 함
    • 변수의 유형은 DEFAULT expression으로 추론됨

• 주의점 및 특징

  • 변수의 선언은 다른 절차적 선언문(procedural statuemnets)에서 전에 노출되어야만 함(혹은 BEGIN 블록의 시작점에서)
  • 변수의 이름은 case-insensitive(대소문자 구분 없음 ↔ case-sensitive)
  • 여러 개의 변수가 DECLARE 에서 선언될 수 있지만, 변수 유형이나 expression은 하나만 가능함
  • 현재 블록 또는 포함 블럭에서 이전에 선언한 변수와 같은 이름을 가진 변수를 선언하는 것은 오류가 남
  • 다중 쿼리문의 실행동안 오류 메시지 등 정보를 체크하기 위한 system variable을 사용할 수 있음(링크)

• 코드 예시

DECLARE x INT64;
DECLARE d DATE DEFAULT CURRENT_DATE();
DECLARE x, y, z INT64 DEFAULT 0;

# 테이블 스키마로부터 item 변수의 유형을 추론하여 default expression과 type 적용  
DECLARE item DEFAULT (SELECT item FROM schema1.products LIMIT 1);

2. SET

• 선언한 변수에 대한 설정값

  • 값을 가질 수 있는 변수를 설정하고(디폴트 말고 초기값 셋팅임), 다중 표현식의 결과에 기반하여 다중 변수를 셋팅함

• 코드 예시

    # x변수는 5로 설정
    SET x = 5;
  
    # a = 4, b = 'foo', c = false 로 설정됨 
    SET (a, b, c) = (1 + 3, 'foo', false);

• 다중 변수에 쿼리의 결과를 할당하는 예시
  1. 두 변수를 선언 (target_word, corpus_count)
  2. SELECT AS STRUCT 쿼리의 결과로 선언한 두 변수에 할당함
     • 쿼리의 결과는 두 필드를 가진 STRUCT의 단일값을 포함
  3. 첫번째 요소는 첫번째 변수에, 두번째 요소는 두번째 변수에 할당함

DECLARE target_word STRING DEFAULT 'methinks';
DECLARE corpus_count, word_count INT64;

SET (corpus_count, word_count) = (
  SELECT AS STRUCT COUNT(DISTINCT corpus), SUM(word_count)
  FROM bigquery-public-data.samples.shakespeare
  WHERE LOWER(word) = target_word
);

SELECT
  FORMAT('Found %d occurrences of "%s" across %d Shakespeare works',
         word_count, target_word, corpus_count) AS result;

• 결과

  • Found 151 occurrences of "methinks" across 38 Shakespeare works

    
    

3. EXECUTE IMMEDIATE

• 동적 SQL문을 즉시 실행시킴

• Syntax

    EXECUTE IMMEDIATE sql_expression [ INTO variable[, ...] ] [ USING identifier[, ...] ];
  
    sql_expression:
      { "query_statement" | expression("query_statement") }
  
    identifier:
      { variable | value } [ AS alias ]

  • sql_expression : 쿼리문 표현문. 단일 DDL or DML 구문. IF와 같은 컨트롤 구문은 사용할 수 없음
  • expression : 함수, 조건식 혹은 서브쿼리의 표현식임
  • query_statement
    • 실행하기 위한 standalone SQL 구문
    • 만약 값을 리턴한다면, INTO 절은 반드시 같은 유형의 값을 포함해야함
  • INTO
    • SQL 표현식이 실행된 이후에, 한 개 또는 그 이상의 변수에서 결과를 저장할 수 있음
  • USING
    • SQL 표현식을 실행하기 전에, USING 절의 한 개 또는 그 이상의 구분자를 쿼리로 전달할 수 있음
    • 이 식별자는 쿼리 파라미터와 유사하게 작동하여 쿼리문에 노출됨
    • 하나의 구분자는 variable 혹은 value가 될 수 있음

• 예제

  • ? : palaceholder(자리표시자)로 USING 절에 인덱스로 구분자와 묶임

    ```sql

  • • y = 1 * (3 + 2) = 5 EXECUTE IMMEDIATE "SELECT ? * (? + 2)" INTO y USING 1, 3;

  • @identifier: USING절에 이름으로부터 구분자와 묶임

    ```sql

  • • y = 1 * (3 + 2) = 5 EXECUTE IMMEDIATE "SELECT @a * (@b + 2)" INTO y USING 1 as a, 3 as b;

• 예제2

    -- create some variables
    DECLARE book_name STRING DEFAULT 'Ulysses';
    DECLARE book_year INT64 DEFAULT 1922;
    DECLARE first_date INT64;
  
    -- Create a temporary table called Books.
    EXECUTE IMMEDIATE
      "CREATE TEMP TABLE Books (title STRING, publish_date INT64)";
  
    -- Add a row for Hamlet (less secure)
    EXECUTE IMMEDIATE
      "INSERT INTO Books (title, publish_date) VALUES('Hamlet', 1599)";
  
    -- add a row for Ulysses, using the variables declared and the ? placeholder
    EXECUTE IMMEDIATE
      "INSERT INTO Books (title, publish_date) VALUES(?, ?)"
      USING book_name, book_year;
  
    -- add a row for Emma, using the identifier placeholder
    EXECUTE IMMEDIATE
      "INSERT INTO Books (title, publish_date) VALUES(@name, @year)"
      USING 1815 as year, "Emma" as name;
  
    -- add a row for Middlemarch, using an expression
    EXECUTE IMMEDIATE
      CONCAT(
        "INSERT INTO Books (title, publish_date)", "VALUES('Middlemarch', 1871)"
      );
  
    -- save the publish date of the first book, Hamlet, to a variable called
    -- first_date
    EXECUTE IMMEDIATE "SELECT publish_date FROM Books LIMIT 1" INTO first_date;
  
    +------------------+------------------+
    | title            | publish_date     |
    +------------------+------------------+
    | Hamlet           | 1599             |
    | Ulysses          | 1922             |
    | Emma             | 1815             |
    | Middlemarch      | 1871             |
    +------------------+------------------+

4. BEGIN...END

BEGIN
  sql_statement_list
END;

• BEGIN : END가 나올 때까지만 선언된 변수가 존재하는 곳의 구문의 블록을 초기화한다.

• sql_statement_list : 0~그 이상의 ; 으로 끝나는 SQL문의 리스트

• 선언된 변수

  • 변수 선언은 구문들 증 다른 유형의 변수 이전에 있는 블록의 시작점에서 나옴

  • 블록 안에서 선언된 변수들은 그 블록과 포함되고 있는 블록 안에 참조될 수 있음

    • 즉 , BEGIN…END 구문 안에서 선언된 변수 이 안에서만 활용 가능하고, 구문 밖에서는 선언이 안됨 (하지만 같은 변수명으로 선언할 수 없음)

    • 예시

        BEGIN
          DECLARE y INT64;
          DECLARE x INT64 DEFAULT 10;
          SET y = x;
          SELECT y, x;
        END;
        SELECT x;

      

- 같은 블록 혹은 다른 블록에서 선언된 변수와 같은 이름을 선언하는 것은 오류가 남

• 블록당 최대 50개 수준을 포함할 수 있고, BEGIN/END, IF/ELSE/END/IF, WHILE/END와 같은 조건문이 있음
• BEGIN/END 는 포함된 요소들과 같이 동적으로 실행됨으로써 제한됨
• Labels
  • BEGINE…END 구문에는 label을 생성 할 수 있음(링크)

    ```sql
    label_name: BEGIN
      block_statement_list
    END [label_name];
    ```

• 예시

DECLARE x INT64 DEFAULT 10;
BEGIN
  DECLARE y INT64;
  SET y = x;
  SELECT y;
END;
SELECT x;

5. BEGIN...EXCEPTION...END

BEGIN
  sql_statement_list
EXCEPTION WHEN ERROR THEN
  sql_statement_list
END;

• BEGIN … EXCEPTION : 블록안에 있는 구문들을 실행함
  • 블록 내 구문에서 에러가 발생한다면, 블록의 나머지 구문은 무시하고, EXCEPTION절이 실행됨
• EXCEPTION 시스템 변수를 이용하여 에러에 대한 세부 정보를 확인할 수 있음

• 예제

    BEGIN
      declare x string default 'exam' ;
      declare y int64 default 12 ;
      select x + y ; # string + int 로 에러 상황 만들기 
    EXCEPTION WHEN ERROR THEN
      SELECT
        @@error.message,
        @@error.stack_trace,
        @@error.statement_text,
        @@error.formatted_stack_trace;
    END;

• 첫 번째 결과 보기

• 두 번째 결과 보기

  

1. 도커 이미지란?

• commit 으로 하는 방식과 build하는 방식으로 이미지를 생성할 수 있음

• 도커는 레이어드 파일 시스템 기반임

  • 레이어드 파일 시스템이란 ?

• 이미지는 프로세스가 실행되는 파일의 집합(환경)

  • 리눅스라 하면 여러개의 파일이 있으면 이 파일이 전체 다 필요함(이걸 압축해놓은 것임)

• 프로세스는 환경(파일)을 변경할 수 있음

• 이 환경을 저장해서 새로운 이미지를 만듬

• 이미지는 읽기 전용 VS 쓰기 전용 이 두가지로 나뉨

  • 읽기 전용
    • 수정할 수 없는 영역
    • 예시로 우분투 이미지는 내가 수정할 수 없지만 새롭게 추가 수정 삭제할 수 있는 것임

• 베이스 이미지에 깃을 추가해서 새로운 커스텀 이미지를 만들 수 있음

• 도커 이미지 네이밍 규칙

  docker build -t mysname/ubuntu:airflow

• 도커 우분투에 깃 이미지 설치하여 이미지 생성

    docker run -it --name git ubuntu:latest bash 
  
    # apt-get update
    # apt-get install -y git
    # git --version
  
    docker images | grep ubuntu
    docker commit git ubuntu:git
    docker images | grep ubuntu

2. Dockerfile

• 서버에 직접 입력하는 과정을 하나의 파일로 관리

  • 최초부터 어떤 과정을 걸처 프로그램이 설치된 것인지 관리하기 편함

• build로 생성하는 도커 이미지를 만들기

• 도커파일 명령어

| FROM | 기본 이미지 |
| --- | --- |
| RUN | 쉘 명령어 실행 |
| CMD | 컨테이너 기본 실행 명령어(Entrypoint의 인자로 사용) |
| EXPOSE | 오픈되는 포트 정보 |
| ENV | 환경변수 설정 |
| ADD | 파일 또는 디렉토리 추가. URL/ZIP 사용가능 |
| COPY | 파일 또는 디렉토리 추가  |
| ENTRYPOINT | 컨테이너 기본 실행 명령어 |
| VOLUMNE | 외부 마운트 포인트 생성  |
| USER | RUN, CMD, ENTRYPOINT를 실행하는 사용자  |
| WORKDIR | 작업 디렉토리 설정 |
| ARGS | 빌드타임 환경변수 생성 |
| LABEL | key-value 데이터 |
| ONBUILD | 다른 빌드의 베이스로 사용될 떄 사용하는 명령어 |

(1) 이미지 빌드하기

# docker build -t {이미지명:이미지태그} {빌드 컨텍스트}
docker build -t sample:1

• 현재 디렉토리의 Dockerfile로 빌드
  • -f <Dockerfile 위치> 옵션을 사용하여 다른 위치의 도커 파일 사용 가능
  • -t 명령어로 도커 이미지 이름을 지정함
  • {네임스페이스}/{이미지명}:{태그} 형식

• 마지막에는 빌드 컨텍스트 위치(경로)를 지정
  • 현재 디렉터리를 의미하는 곳을 주로 사용 → .
  • 필요한 경우 다른 디렉터리를 지정할 수 있음

• [참고] .dockerignore
  • .gitignore 과 비슷한 역할
  • 도커 빌드 컨텍스트에서 지정된 패턴의 파일을 무시함
  • .git이나 민감한 정보를 제외하는 용도로 주로 사용
  • .git이나 에셋 디렉터리만 제외시켜도 빌드 속도 개선
  • 이미지 빌드 시에 사용하는 파일은 제외시키면 안 됨

(2) Dockerfile 생성

FROM ubuntu:latest # 이미지명을  
RUN apt-get update # 
RUN apt-get install -y git # 

(3) build 명령어를 이용한 이미지 빌드

docker build -t ubuntu:git-dockerfile .
docker images | grep ubuntu

• 도커 컴포즈는 여러 컨테이너를 실행하고 정의하기위한 툴
• 어플리케이션 서비스를 구성하기 위한 YAML파일을 사용함
• 컴포즈는 모든 환경에서 작동한다
  • 제작
  • 스테이징
  • 개발
  • 검증
  • CI workflow

• 모든 앱의 생명주기를 관리하기위해서 명령어를 가지고 있음
  • start, stop, rebuild 서비스
    • 서비스 = 컨테이너의 애플리케이션
  • 실행하고 있는 서비스의 상태를 봄
  • 실행하고 있는 서비스들의 로그 결과를 스트리밍함
    • 로그를 이용하여 컨테이너별 로그를 확인하는 명령어를 수행할 필요가 없음
  • 하나의 서비스의 일회성(one-off) 명령어를 실행함
    • 여러 컨테이너로 묶어서 실행된다 하더라도 경우에 따라서는 단일 컨테이너에 명령을 내려야 하는 경우도 있는데, 이를 가능하게 함

• 도커 컴포즈의 주요한 특징(장점)
  • 단일 호스트에서 여러 격리된 환경을 가짐
  • 컨테이너 생성시 볼륨 데이터를 보존함
  • 변화한 컨테이너만 재생성할 수 있음
  • 환경 간에서 구성이 변화하는 것을 지원하고 변수를 지원함

• 기본적으로 Docker for mac or window를 설치하면 기본적으로 docker compose가 설치됨

• Linux의 경우는 docker-compose를 따로 설치해줘야 함

    sudo curl -L "https://github.com/docker/compose/releases/download/1.26.0/docker-compose-$(uname -s)-$(uname -m)"
    sudo chmod +x /usr/local/bin/docker-compose
  
    # or
    sudo curl -L "https://github.com/docker/compose/releases/download/1.26.0/docker-compose-$(uname -s)"
    sudo curl -L "https://github.com/docker/compose/releases/download/1.26.0/docker-compose-Linux-x86_64"

• 설치 확인하는 방법

    docker-compose version

2. Yaml이란?

• 구성 파일 작성에 자주 사용되는 데이터 직렬화 언어
  • 직렬화란?(or Serialization)
    • Data Science의 데이터 스토리지 문맥에서 데이터 구조나 오브젝트 상태를 동일하거나 다른 컴퓨터 환경에 저장하고 나중에 재구성할 수 있는 포맷으로 변환하는 과정
    • 즉, 객체에 저장된 데이터를 스트림(데이터의 흐름)에 쓰기 위해 연속적인 데이터를 변환하는 것
      • 데이터 스트림이란 로드된 데이터를 의미함
      • 데이터 스트림은 외부(서버)에서 받아 온(로드된 또는 스트리밍 된) 데이터(이미지, 동영상 등의 멀티미디어)를 의미
    • 한 번의 읽기 또는 쓰기 동작으로 전송되는 정보
    • 진짜 데이터가 열을 지어 차례대로 입력되는 것을 의미

• .yml 또는 .yaml 이라는 확장자를 가짐

• 띄어쓰기를 가지고 데이터를 표현(탭 문자는 허용되지 않음)

• YAML은 Yet Another Markup Language 혹은 YAML ain’t markup language(재귀 약어)

  • 후자의 경우 YAML은 문서가 아닌 데이터용임을 강조
  • 사람이 읽을 수 있고 이해하기도 쉬어 프로그래밍 언어 중에 인기가 높음
  • 다른 프로그래밍 언어와 함께 사용할 수 있음

3. 도커 컴포즈로 이미지 생성 및 실행하기

• 실무에서는 대부분 YAML파일로 작업해서 하는 편임
• 클라이언트를 기반으로 한땀한땀 명령어를 입력하고 다시 입력하고 이런 반복적인 작업을 표현하는 방식이 깔끔하지 못함 → Yaml파일 작성하여 docker-compose로 실행함

_* 아래 예시는 인프런 강의 예제를 참고함 _ ** [ 예시 ] **

1. Yaml파일 생성
    - Client로 한땀한땀 입력한 코드를 한번에 key value형태로 작성

    ```bash
    version: '2'
    services:
      db: # name 
        image: mariadb:10.9 # docker 명령어 제일 마지막에 썼던 이미지와 태그
        volumes: # -v 옵션으로 주었던 데이터 유실을 방지하기 위한 옵션 
          - ./mysql:/var/lib/mysql
        restart: always # 컨테이너가 만약 죽게되면 도커가 자동으로 띄어주는 역할(항상 재시작) 
        environment: # -e 로 지정했던 환경변수를 지정해줌 
          MYSQL_ROOT_PASSWORD: wordpress
          MYSQL_DATABASE: wordpress
          MYSQL_USER: wordpress
          MYSQL_PASSWORD: wordpress
      wordpress:
        image: wordpress:latest
        volumes:
          - ./wp:/var/www/html
        ports:
          - "8000:80"
        restart: always
        environment:
          WORDPRESS_DB_HOST: db:3306
          WORDPRESS_DB_USER: wordpress
          WORDPRESS_DB_PASSWORD: wordpress
    ```

2. `docker-compose`로 실행
    - 자동으로 docker-compose.yml 파일을 읽어서 실행함

[ 참고 자료 ]

• overview - docker compose

• YAML 개념, 기능, 특징: YAML 파일 및 포맷, 구문 예시, 사용법

• 인프런 강의 '초보를 위한 도커 안내서'

시작하기 앞서, 준비!

• Docker Commuitiy Edition 설치(최소 4GB memory 이상)
• Docker compose v1.29.1 이상 혹은 최신 버전 설치

1. docker-compose.yaml 파일 다운로드

• Docker compose로 Airflow를 개발하기 위해서는 docker-compose.yaml파일을 설치해야함
  • curl이라는 명령어는 프로토콜들을 이용해 URL 로 데이터를 전송하여 서버에 데이터를 보내거나 가져올때 사용하기 위한 명령줄 도구 및 라이브러리임

curl -LfO 'https://airflow.apache.org/docs/apache-airflow/2.5.1/docker-compose.yaml'

• 설치가 완료되면 아래와 같이 docker-compose.yaml 파일이 생성됨

• docker-compose.yaml 내 서비스

  • airflow-scheduler : 모든 task와 DAG를 모니터링

  • airflow-webserver : 에어플로우 웹 서버. http://localhost:8080

  • airflow-worker : 스케줄러로부터 부여된 테스크들을 실행하기 위한 워커

  • airflow-init : 서비스 이니셜라이제이션

  • postgres(DB) : 데이터 베이스

  • redis : 스케줄러에서 워커로 메시지를 보내는 브로커 * 서비스는 Celery Executor로 실행됨

    _(참고) JSON형태가 익숙하다면, YAML -> JSON 포맷으로 변환하여 확인하면 좀 더 이해하기 쉬움 (링크) _

2. 도커 컨테이너 띄우기

• docker compose up 코드로 멀티 컨테이너 서비스 띄우기

```bash
docker-compose up 
```

- `docker ps` 명령어로 현재 띄어진 서비스별 컨테이너를 확인할 수 있음 
![](https://velog.velcdn.com/images/sophi_e/post/d76d362c-95c9-4c31-8da6-c9bb37f93afb/image.png)

• 컨테이너의 폴더는 마운트됨 → 컴퓨터와 컨테이너간의 일치 (yaml파일 내 volume 항목 안에 각각의 directory를 매치해주면 됨)

  • ./dags : Dag file 디렉토리
  • ./logs : task 실행 과 스케줄러의 로그 파일
  • ./plugins : 커스텀 플러그인(링크)

• Option

  • flower앱 컨테이너도 함께 사용하기 (--profile flower 추가)

    • 환경을 모니터링함

    • http://localhost:5555

      docker compose --profile flower up 
      혹은 
      docker compose up flower

3. Initializing Environment

• 에어플로우 개발 전 환경 구성과 유저 권한을 셋팅해야함

(1) 에어플로우 유저 권한 셋팅

mkdir -p ./dags ./logs ./plugins # mkdir -p 옵션 : 여러 하위 디렉토리를 생성시에 사용한다.
echo -e "AIRFLOW_UID=$(id -u)" > .env

AIRFLOW_UID=50000

• 유저 권한 입력이 안될 시 이런 WARN이 뜸

• (참고) docker compose 지원 환경변수

(2) 에어플로우 DB초기화

docker-compose up airflow-init

(3) CLI RUN

• Airflow command를 실행시키기 위해서는 두가지 방법이 있음
  1. CLI에 docker-compose run airflow-worker airflow info 입력

     curl -LfO 'https://airflow.apache.org/docs/apache-airflow/2.5.1/airflow.sh'
     chmod +x airflow.sh

2. Mac 또는 Linux의 경우에는 아래 코드 실행 후 sh파일로 실행시킴

   curl -LfO 'https://airflow.apache.org/docs/apache-airflow/2.5.1/airflow.sh'
   chmod +x airflow.sh

   

   위 코드 실행시 airflow.sh 파일이 생성됨

   ./airflow.sh airflow info

   • 1, 2번 모두 실행시 아래와 같은 동일한 결과를 얻을 수 있음

(4) Webserver 접속하기

• 디폴트로 설정된 http://localhost:8080 링크로 접속

• 디폴트 계정과 비밀번호인 airflow로 접속함

• 풀 목록 검색 요청 curl

ENDPOINT_URL="http://localhost:8080/"
curl -X GET  \
    --user "airflow:airflow" \
    "${ENDPOINT_URL}/api/v1/pools"

• 잘못된 정보를 입력했을 때 → Unauthorized로 에러 메시지를 보냄

4. 컨테이너 stop or delete image

docker compose down --volumes --rmi all

5. yaml&Dockerfile로 이미지 커스텀하는 방법

• 확장된 이미지를 사용할 때(새로운 파이썬 패키지 추가, airflow provider 최신 버전 업그레이드 등) 이미지를 커스텀할 수 있음
• 아래 예시는 에어플로우 개발을 위한 docker-compose.yaml 파일에 airflow-code-editor를 설치하기 위한 과정을 정리함

1. docker-compose.yaml 파일 내 build : . 명령어를 추가

   • 기본적으로 ymal파일 내의 build 명령어는 주석처리가 되어있음 -> 해제

     

- 주석을 제거함

    ![](https://velog.velcdn.com/images/sophi_e/post/543c2bd5-ab30-431b-98be-149b3163ed26/image.png)


2. **Dockerfile**을 `docker-compose.yaml` 파일 위치에 추가 
    - airflow-code-editor 패키지를 설치하는 명령어를 `Dockerfile` 내에 기재 ([명령어 참고 문서](https://airflow.apache.org/docs/docker-stack/build.html))

    ```bash
    FROM apache/airflow:2.5.1-python3.8
    RUN pip install airflow-code-editor
    ```

3.  두 파일이 있는 위치에서 `docker-compose build`  명령어를 통해 이미지를 빌드 

   - 기존 에어플로우와 Dockerfile을 이용하여 airflow-code-editior 패키지를 설치하여 이미지를 생성

    ```bash
    docker-compose build
    ```
    ![](https://velog.velcdn.com/images/sophi_e/post/0294d893-4f99-4749-bec8-b90d830c46d2/image.png)

4. 빌드된 도커 이미지를 실행 

```bash
docker-compose up 
```
![](https://velog.velcdn.com/images/sophi_e/post/75fe4b5d-abb0-404c-8743-6f7f7095129c/image.png)


5. 결과 확인 

 - 기본 웹서버에 존재하지 않는 DAGs Code Editor 생성됨

1. 등장 배경

• 전가상이든 반가상이든 추가적인 OS를 설치하여 가상화하는 방법은 어쨋든 성능문제가 있었고 이를 개선하기 위해 프로세스를 격리하는 방식이 등장함

• 즉, 서버를 관리한다는 건 굉장히 까다로운 문제임
  • 하나의 서버에 여러개의 프로그램을 설치하는것 → 까다로움
  • 사용하는 라이브러리의 버전이 다르거나 동일한 포트를 사용하는 경우, 설치가 굉장히 까다로움 → 차라리 다른 서버에 설치하는게 나았고, 조립 PC가 늘어나고 자원은 낭비됨

• DevOps의 등장으로 개발주기가 짧아지면서 배포는 자주 이루어지고 마이크로서비스 아키텍쳐가 유행하면서 프로그램은 더 잘 쪼개어져 관리는 더 복잡해짐
  • DevOps란?
    • DevOps is a set of practices that combines software development (Dev) and IT operations (Ops).
    • It aims to shorten the systems development life cycle and provide continuous delivery with high software quality.
    • DevOps is complementary with Agile software development; several DevOps aspects came from the Agile methodology.

→ 따라서, 새로운 툴은 계속 나오고 클라우드 발전으로 설치해야 할 서버가 수백, 수천대에 이르는 상황에서 도커가 등장하고 서버관리 방식은 완전히 바뀌게 됨

2. 도커란?

• 서버의 컨테이너도 흔히 아는 컨테이너와 비슷하게 다양한 프로그램, 실행환경을 컨테이너로 추상화하고 동일한 인터페이스를 제공하여 프로그램의 배포 및 관리를 단순하게 해줌

• 백엔드 프로그램, 데이터베이스 서버, 메시지 큐 등 어떤 프로그램도 추상화할 수 있음

• 서버에서 다양한 프로그램, 실행 환경을 컨테이너로 추상화하고 동일한 인터페이스를 제공하여 프로그램 배포 및 관리를 단순하게 해줌

• 백엔드 프로그램 데이터베이스 서버, 메시지 큐등 어떤 프로그램도 콘테이너로 추상화될도 있고 조립PC, AWS, Azure, Google cloud등 어디서든 실행할 수 있음

3. 컨테이너란?

• 격리된 공간에서 프로세스가 동작하는 기술임
• 기존 방식
  • 가상화 방식(주로 OS 가상화)
  • VMware나 Virtual Box 같은 가상머신은 호스트 OS위에 게스트 OS 전체를 가상화하여 사용하는 방식
    • 이 방식은 여러가지 OS를 가상화 할 수 있고 비교적 사용법이 간단하지만, 무겁고 느려서 운영환경에선 사용할 수 없음
• 개선 방식
  • CPU 가상화 기술(HVM)을 이용한 KVM(Kernel-based Virtual Machine)과 반가상화(Paravirtualization) 방식의 Xen이 등장
  • 이런 방식은 게스트 OS가 필요하긴하지만 전체 OS를 가상화하는 방식이 아니였기 때문에 호스트형 가상화 방식에 비해 성능이 향상됨
  • 이러한 기술들은 OpenStack이나 AWS, Rackspace같은 클라우드 서비스에서 가상 컴퓨팅 기술의 기반이 되었음

4. 이미지란?

• 컨테이너를 실행하기 위한 모든 정보를 가지고 있음
  • 의존성 파일을 컴파일하고 이것저것 설치할 필요가 없음

• 새로운 서버가 추가되면 미리 만들어 놓은 이미지를 다운받고 컨테이너를 생성하기만 하면 됨

• 한 서버에 여러개의 컨테이너를 실행할 수 있고 수십, 수백, 수천대의 서버도 문제없음

• 도커 이미지는 Docker Hub나 Docker Registry 저장소를 직접 만들어 관리할 수 있음

5. 도커가 핫한 이유?

이미지 레이어 구조로 저장방식으로 도커가 관리하는 파일 시스템 영역에 이미지를 복사함

• 도커는 완전히 새로운 기술이 아니며 이미 존재하는 기술을 잘 포장함

• 도커의 이미지를 이용하여 docker run을 하면 도커는 도커가 관리하는 파일 시스템 영역에 이미지를 복사함
  • 복사후, 도커는 이미지의 최상단에 컨테이너 레이어라고 불리는 얇은 레이어를 추가하여 컨테이너를 생성함
  • 그리고 사용자에게 유니온 파일 시스템을 이용하여 마치 이러한 여러개의 파일 시스템(Image Lager)로 구성되어 있는 이미지 스택 구조가 하나의 파일 시스템 처럼 보이도록 하게 함

• 사용자가 컨테이너 안에서 읽고 쓰는 모든 작업들은 이 컨테이너 레이어에 기록되고 이미지 레이어에는 도지 않음
  • Image layer는 변경 불가(Read only layer)
  • Container layer는 변경 가능(Readable/Writable layer)

• 이미지 레이어는 삭제되지 않지만 컨테이너 레이어는 해당 컨테이너가 종료될 경우 같이 소멸됨
  • 사용자가 컨테이너 레이어에서 작업하던 내용을 이미지 레이어로 적용하고 싶을 경우, docker commit 을 통해 기존 Image layer들과 직접 작성한 컨테이너 레이어를 이용해 새로운 이미지를 생성할 수 있음

6. 특징

네트워크 성능이 중요한 프로그램의 경우 —net=host 옵션을 고려해야 함 참고로, 도커의 기본 네트워크 모드는 Bridge 모드로 약간의 성능 손실이 있음

• 하나의 서버에 여러개 컨테이너를 실행하면 서로 영향을 미치지 않고 독립적으로 실행되어 마치 가벼운 VM(Virtual Machine)을 사용하는 느낌을 줌

• 실행중인 컨테이너에 접속하여 명령어를 입력할 수 있고 apt-get 이나 yum 으로 패키지를 설치할 수 있음

• 사용자 추가하고 여러개의 프로세스를 백그라운드로 실행할 수 있음

• CPU나 메모리 사용량을 제한할 수 있고 호스트의 특정 포트와 연결하거나 호스트의 특정 내부 디렉토리인 것처럼 사용할 수 있음

• 새로운 컨테이너를 만드는데 걸리는 시간은 겨우 1-2초로 가상머신과 비교할 수 없이 빠름

참고 자료

• 초보를 위한 도커 안내서 - 도커란 무엇인가?
  • https://subicura.com/2017/01/19/docker-guide-for-beginners-1.html
• 인프런 강의 - 초보를 위한 도커 안내서
  • https://www.inflearn.com/course/%EB%8F%84%EC%BB%A4-%EC%9E%85%EB%AC%B8/dashboard

Python에서 조건에 따른 값 할당을 어떻게 하는지 크게 5가지 방법으로 나누어 정리하였습니다.

세부적으로 정리하자면, binary or multiple condition에 따라서 사용하는 방식을 나누어 보았습니다.

1. binary condition
   • np.where() function
   • lambda 형식

2. multiple condition
   • np.select()
   • pd.apply()
   • df.loc()

** 예제 데이터는 아래와 같습니다.

1. numpy.where() 함수

• ‘True’ or ‘False’과 같은 이진 조건의 기반이 되는 새로운 컬럼을 추가할 때 유용한 함수임

• np.where(condition, value if condition is true, value if condition is false)

• 예시 코드

  import pandas as pd
  import numpy as np
  dfTest['분기'] = np.where(dfTest['월'] <= 6, '상반기', '하반기')

2. lambda 형식

• 람다 함수 또한 이진 조건에 유용한 함수임
  • (참고) 람다 표현식 : https://wikidocs.net/64

• df['new column name'] = df['column name'].apply(lambda x: 'value if condition is true' if x condition else 'value if condition is false')

• 예제 코드

    dfTest['분기'] = dfTest['월'].apply(lambda x: '상반기' if x <= 6 else '하반기')

3. numpy.select() function

• 다양한 조건을 포함하는 카테고리 컬럼을 만들 때

• condition 대한 리스트 생성 -> 조건에 해당하는 value 리스트 생성 -> np.select() 함수를 이용하여 적용

  1. 순서대로 조건을 명시하여 리스트를 생성하기

     #Create a python list to specify all the conditions
     #Create a python list to specify all the conditions
     conditions = [ 
        (dfTest['월'] == 12) | (dfTest['월'] <= 2),
        (dfTest['월'] >= 3) & (dfTest['월'] <= 5),
        (dfTest['월'] >= 6) & (dfTest['월'] <= 8),
        (dfTest['월'] >= 9) & (dfTest['월'] <= 11)
        ]
     

  2. 위 순서의 각 조건에 해당하는 값(’label’)을 할당하여 파이썬 리스트로 생성

     #create a python list of values to be assigned to the conditions
     values = ['겨울', '봄', '여름', '가을']

  3. np.select() 함수를 사용하여 새로운 컬럼 생성

  • 생성한 두 개의 리스트들을 arguments로 사용하여 np.select를 이용하여 새로운 컬럼으로 생성

    # use np.select() to create a new column
    dfTest['계절'] = np.select(conditions, values)

    

(참고) 1,2,3을 한 번에 적용할 때 

```python
df['visits_category'] = np.select([
    (df['visits_30days'] == 0),
    (df['visits_30days'] > 0) & (df['visits_30days'] <= 5),
    (df['visits_30days'] > 5) & (df['visits_30days'] <= 10),
    (df['visits_30days'] > 10)
    ], ['0 visits', '1-5 visits', '6-10 visits', '>10 visits'])
```

4. pandas apply 함수

• 조건을 특정하는 함수를 생성한 후 → 컬럼에 적용

  1. 함수 생성

     def conditions(row):
      if (row['월'] == 12 ) | (row['월'] <= 2) :
          val = '겨울'
      elif (row['월'] >= 3) & (row['월'] <= 5 ):
          val = '봄'
      elif (row['월'] >= 6) & (row['월'] <= 8) :
          val = '여름'   
      elif (row['월'] >= 9) & (row['월'] <= 11) :
          val = '가을'   
      else:
          val = 'NA'
      return val

  2. 컬럼에 함수 적용

     
     #Apply the function to each data point in the data frame
     dfTest['계절2']= dfTest.apply(conditions, axis=1)

     

5. DataFame loc function() 활용

• np.select와 비슷한 원리
• 데이터프레임의 loc함수를 통해 특정 조건의 row에 새로운 column을 부여하여 값을 할당하는 방식
• df.loc[기존 행의 조건 입력, ‘새로운 열’] = ‘label’

dfTest.loc[(dfTest['월'] == 12) | (dfTest['월'] <= 2), '계절3'] = '겨울' 
dfTest.loc[(dfTest['월'] >= 3) & (dfTest['월'] <= 5), '계절3'] = '봄' 
dfTest.loc[(dfTest['월'] >= 6) & (dfTest['월'] <= 8), '계절3'] = '여름'  
dfTest.loc[(dfTest['월'] >= 9) & (dfTest['월'] <= 11) , '계절3'] = '가을'

[자료 링크]

• 5 Ways to Apply If-Else Conditional Statements in Pandas

원인1. DAG run timeout

• dagrun_timeout 값이 낮은 원인
• 타임아웃이 나기 전에 Dag Run이 얼마나 오래 걸리는지에 대한 정의를 보통 함
• 따라서 타임아웃이 예정된 DagRun에 영향을 받는지 확인해봐야함
• 해결 방법
  • Airflow UI로 Failed DagRun duration 확인하여 Dag파일(.py)의 dagrun_timeout을 그 이상의 값으로 설정
    • 아래 dagrun_timeout 의 value 수정
      

원인2. Running out of memory

• OOM(Out of memory) 현상으로 발생되는 원인일 수 있음
• 에어플로우의 개발 환경에서 메모리가 충분하지 않으면 발생할 수 있는 에러로 메모리 사용량을 높이는 요소 조정
• 해결 방법
  • max_active_tasks_per_dag 의 value 조정
  • parallelism 의 value 조정

원인3. Metadata Database draining the CPU

• 메타데이터 데이터베이스의 CPU사용량이 100%인 경우

• 해결 방법

  • job_heartbeat_sec 의 값을 조정해야함

  • airflow.cfg의 job_heartbeat_sec 수정 (default : 5 seconds)

    vim airflow.cfg
    
    # 아래 값 확인 후 변경 
    [scheduler]
    job_heartbeat_sec = 30 

    

원인4. Disable “Mini Scheduler”

• 기본적으로 퍼포먼스를 개선하기 위해서 task supervisor process가 "미니 스케줄러"를 수행하여 동일한 DAG의 더 많은 작업을 예약함

• 이 설정을 그대로 두면 동일한 DAG의 작업이 더 빨리 실행되지만 경우에 따라 다른 dag에 영향을 미침

• 해결방법

  • airflow.cfg의 schedule_after_task_execution 수정 (default : True)

    [scheduler]
    schedule_after_task_execution = False

    

• 참고 자료
  • How To Fix Task received SIGTERM signal In Airflow

* 참고 : Apache Airflow 기반의 데이터 파이프라인 도서, Airflow documentation

1. Airflow란?

• Apache Airflow는 초기 에어비엔비(Airfbnb) 엔지니어링 팀에서 개발한 워크플로우 오픈 소스 플랫폼 _ ** 워크플로우란? : 의존성으로 연결된 작업(Task)들의 집합_

               (ex) ETL의 경우 Extractaction > Transformation > Loading 의 작업의 흐름 

• 프로그래밍 방식으로 워크플로우를 작성, 예약 및 모니터링

2. Airflow 기본 구성 및 작동 원리

(1) Airflow Key Concept

a. DAG (Directed Acyclic Graph)

• 단어 뜻 그대로 순환하지 않는 그래프, DAG(대그)라고 부름
• 반복이나 순환을 허용하지 않음
• 순차적으로 작업(task)이 이루어지며, 순환 실행을 방지하기 때문에 매우 중요함 → 논리적 오류는 교착상태(deadlock)로 이어짐

b. Operator

• Operator(Task)를 정의하는데 사용

• Operator Type

  • Action Operators
    • 기능이나 명령을 실행하는 오퍼레이터
    • 실제 연산을 수행, 데이터 추출 및 프로세싱
    • (참고) 내장 Operators는 BashOperator, PythonOperator, EmailOperator
      • 이외의 오퍼레이터는 공식 documnet 참고 (링크)
  • Transfer Operater
    • 하나의 시스템을 다른 시스템으로 옮김 (데이터를 Source에서 Destination으로 전송 등 )
    • 예를 들어, Presto에서 MySQL로 데이터를 전송하는데에 사용
  • Sensor Operators
    • 조건이 만족할 때까지 기다렸다가, 조건이 충족되면 다음 Task를 실행시킴

      c. Task & Task Instance

• Task : 데이터 파이프라인에 존재하는 Operator를 의미

  • Operator를 실행하면 Task가 됨

• Task Instance

  • 데이터 파이프 라인이 Trigger되어 실행될 때 생성된 Task를 Task Instance라고 함

  • 태스크 실행 순서 정의

    • 오른쪽 시프트 연산자(binary right shift operator), 즉 rshift(>>)를 사용하여 태스크의 의존성 정의함

      Task1 >> Task2 >> Task3 
      [Task1, Task2] >> Task3 

• Task VS Operator

  • 사용자 관점에서는 두 용어를 같은 의미지만 Task 는 작업의 올바른 실행을 보장하기 위한 Manager임
  • 사용자는 Operator를 사용해서 수행할 작업에 집중하며, Airflow는 태스크를 통해 작업을 올바르게 실행함

  → 사용자는 각 환경별 작업이 잘 이루어지는지 확인하기 위해서 Operator내 코드를 구성하는데에 집중하고, Airflow는 각 Operator 내의 구성 요소들이 전부 잘 맞아야 작업이 이루어지는 형태의 차이로 보임

(2) Airflow component

• 에어플로우는 웹서버, 스케줄러, metastore, Executor, Worker로 크게 5개의 기본 구성으로 이루어져 있음

  • 웹서버 : 웹 대시보드 UI로 스케줄러에서 분석한 Dag를 시각화하고 DAG 실행과 결과를 확인할 수 있는 인터페이스를 제공함
  • 스케줄러 : DAG를 분석하고 현재 시점에서 Dag의 스케줄이 지난 경우 airflow worker에 DAG의 태스크를 예약함
  • Worker : 예약된 태스크를 실제로 실행시키는 것
  • Metastore : 에어플로우에 있는 Dag, Task등의 메타데이터 관리
  • Executor : 태스크가 어떻게 실행되는지 정의

(출처 : Apache Airflow 기반의 데이터 파이프라인)

(3) Airflow 기본 동작 원리

1. 유저가 새로운 Dag를 작성 → Dags Foolder 안에 py 파일 배치
2. Web Server와 Scheuler가 파싱하여 읽어옴
3. Scheduler가 Metastore를 통해 DagRun 오브젝터를 생성함
   1. DagRun은 사용자가 작성한 Dag의 인스턴스임 DagRun Status : Running
4. 스케줄러는 Task Instance Object를 스케줄링함
   1. Dag Run object의 인스턴스임
5. 트리거가 상황이 맞으면 Scheduler가 Task Instance를 Executor로 보냄
6. Exeutor는 Task Instance를 실행시킴
7. 완료 후 → MetaStore에 완료했다고 보고함
   1. 완료된 Task Instance는 Dag Run에 업데이트됨
   2. Scheduler는 Dag 실행이 완료되었는지 Metastore를 통해 확인 후에 Dag Run의 생태를 완료로 바꿈 DagRun Status : Completed
8. Metastore가 Webserver에 업데이트해서 사용자도 확인

3. Airflow 장단점

(1) 장점

• 파이썬 코드를 이용하여 파이프라인을 구현하므로 파이썬 언어에서 구현할 수 있는 대부분의 방법을 사용하여 복잡한 커스텀 파이프라인을 만들 수 있음
  • 확장성 - 파이썬 기반으로 쉽게 확장 가능하고 다양한 시스템과 통합이 가능
    • 다양한 유형의 DB, Cloud Service 등 통합 가능
• 데이터 인프라 관리, 데이터 웨어하우스 구축, 머신러닝/분석/실험에 데이터 환경 구성에 유용함
• 에어플로우의 스케줄링 기능으로 DAG에 정의된 특정 시점에 트리거할 수 있을 뿐만 아니라 최종 시점과 예상되는 다음 스케줄 주기를 상세하게 알려줌
• 백필 기능을 사용하면 과거 데이터를 손쉽게 재처리할 수 있기에 코드를 변경후 재생성이 필요한 데이터 재처리가 가능함

(2) 단점

• 파이썬 경험이 없는 경우 DAG 구성의 어려움이 있을 수 있음
• 초기 설치는 간단해 보일지라도 작은 환경 변화에도 작동에 오류가 나는 경우가 있어 롤백하는 경우가 꽤 있음

(3) 주의 사항

• Data Streaming Solution 적용하기엔 적합하지 않음
  • 초단위의(그 이하) 데이터 처리가 필요한 경우에 사용하기에는 부적절함
  • 에어플로우는 반복적이거나 배치 태스크를 실행하는 기능에 초점이 맞춰져 있음
• Data Processing Framework (like Flink, Spark, Hadoop, etc ..)로 사용하는 것은 부적절함
  • 데이터 프로세싱 작업에 최적화 되어있지않아서 매우 느림
  • 경우에 따라 메모리 부족으로 작업이 진행되지 않을 수도 있음 → SparkSubmitOperator와 같은 Operator를 이용하여, 데이터 프로세싱은 Spark와 같은 외부 Framework로 처리
• 파이프라인 규모가 커지면 파이썬 코드가 굉장히 복잡해질 수 있음 → 초기 사용 시점에 엄격한 관리를 해야함