1
2
3
4

environment:
  AIRFLOW_VAR_DATA_DIR: /opt/airflow/data
  # 환경변수가 존재하지 않는다면(세팅되지 않았다면) :- 뒤의 이름들을 해당 변수에 추가하라는 의미
  _PIP_ADDITIONAL_REQUIREMENTS: ${_PIP_ADDITIONAL_REQUIREMENTS:- yfinance pandas numpy oauth2client gspread}

1
2
3
4
5
6
7

x-airflow-common:
  &airflow-common
  ...
  environment:
    &airflow-common-env
    AIRFLOW_VAR_DATA_DIR: /opt/airflow/data
    AIRFLOW_CONN_TEST_ID: test_connection

 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
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52

def get_Redshift_connection():
    hook = PostgresHook(postgres_conn_id = 'redshift_dev_db')
    return hook.get_conn().cursor()

def execSQL(**context):
    schema = context['params']['schema']
    table = context['params']['table']
    select_sql = context['params']['sql']

    ...

    cur = get_Redshift_connection()

    # drop table if exits, before CTAS
    sql = f"""DROP TABLE IF EXISTS {schema}.temp_{table};CREATE TABLE {schema}.temp_{table} AS """
    sql += select_sql
    cur.execute(sql)

    cur.execute(f"""SELECT COUNT(1) FROM {schema}.temp_{table}""")
    count = cur.fetchone()[0]
    if count == 0:
        raise ValueError(f"{schema}.{table} didn't have any record")

    # 중복 레코드, Primary key 등 여러가지 테스트 진행 가능
    # DBT를 사용하면 이런 여러가지 테스트를 직접 작성할 필요 없어짐

    ...

# @once : 주기적으로 실행되지 않고 한번만 실행됨
dag = DAG(
    dag_id = "Build_Summary",
    start_date = datetime(2021,12,10),
    schedule = '@once',
    catchup = False
)

execsql = PythonOperator(
    task_id = 'mau_summary',
    python_callable = execSQL,
    params = {
        'schema' : 'imsolem1226',
        'table': 'mau_summary',
        'sql' : """SELECT
  TO_CHAR(A.ts, 'YYYY-MM') AS month,
  COUNT(DISTINCT B.userid) AS mau
FROM raw_data.session_timestamp A
JOIN raw_data.user_session_channel B ON A.sessionid = B.sessionid
GROUP BY 1
;"""
    },
    dag = dag
)

1
2
3
4
5
6
7
8

# Python dictionary 형태로 사용하기 위해 .py 확장자를 가지게 됨
{
'table': 'mau_summary',
'schema': 'imsolem1226',
'main_sql': """SELECT ...;""",
'input_check': [ ],
'output_check': [ ],
}

1

curl -X POST -H 'Content-type: application/json' --data '{"text":"Hello, Slack!"}' https://hooks.slack.com/services/{slack_webhook_url}