[Apache Airflow] 설치 및 간단 테스트

Airflow 를 정상적으로 사용하기 위해서는 MySQL 과 같은 DB 구성이 되어 있어야 합니다.

(설치 되어 있지 않더라도 기본적인 Hello World 수준의 테스트는 가능 합니다.)

Reference)

https://airflow.apache.org/start.html

$ pip install apache-airflow

or 

$ sudo -H pip install apache-airflow --ignore-installed matplotlib

# initdb 하기 전에 mysql 구성 및 database 생성을 해야 합니다.

$ airflow initdb

$ airflow webserver -p 8080

# MySQL 연결

# The SqlAlchemy connection string to the metadata database.

# SqlAlchemy supports many different database engine, more information

# their website

sql_alchemy_conn = mysql+pymysql://root:000000@localhost/airflow

# MySQL Connection URL Syntax

https://dev.mysql.com/doc/connector-j/8.0/en/connector-j-reference-jdbc-url-format.html

protocol//[hosts][/database][?properties]

# Database 와 Table 에 대한 charset 을 아래와 같이 생성해 줘야 합니다.

ALTER DATABASE database_name CHARACTER SET = utf8mb4 COLLATE = utf8mb4_unicode_ci;

ALTER TABLE table_name CONVERT TO CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci;

# 터미널에서 실행하고 결과 확인 하기

$ airflow test airflow_tutorial_v01 print_hello 2018-05-08

import datetime as dt

from airflow import DAG

from airflow.operators.bash_operator import BashOperator

from airflow.operators.python_operator import PythonOperator

def print_world():

    print('world')

default_args = {

    'owner': 'me',

    'start_date': dt.datetime(2017, 6, 1),

    'retries': 1,

    'retry_delay': dt.timedelta(minutes=5),

}

with DAG('airflow_tutorial_v01',

         default_args=default_args,

         schedule_interval='0 * * * *',

         ) as dag:

    print_hello = BashOperator(task_id='print_hello',

                               bash_command='echo "hello"')

    sleep = BashOperator(task_id='sleep',

                         bash_command='sleep 5')

    print_world = PythonOperator(task_id='print_world',

                                 python_callable=print_world)

print_hello >> sleep >> print_world

일부 테스트가 부족한 부분이 있기 때문에 그냥 참고 정도만 하세요.