[숙제 - 51일 차] mau_summary, channel_summary를 config로 옮기기

mau_summary, channel_summery를 config로 옮기기

아래 링크의 'ELT 구현'을 확인해 보면 nps_summary.py를 예시로 config에 대한 이해를 할 수 있다. nps_summary의 형태를 참고하여 mau_summary, channel_summary를 작성해 보자. nps_summary에서는 input_check, output_check를 통해 테스트가 추가되어 있는데, 해당 부분은 제외하고 필요한 부분만 작성할 것이다.

[TIL - 51일 차] Airflow 고급기능과 DBT, 데이터 디스커버리 (1)

config/mau_summary.py

input_check와 output_check는 사용하지 않을 것이기 때문에 아예 지웠다가 redshift_summary Operator의 __init__을 실행하던 중 에러가 발생하였다. 그래서 빈 리스트 형태로 적어주었다.

{
          'table': 'mau_summary',
          'schema': 'ss721229',
          'main_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;""",
          'input_check': [],
          'output_check': []
}

config/channel_summary.py

{
          'table': 'channel_summary',
          'schema': 'ss721229',
          'main_sql': """
SELECT
  DISTINCT A.userid,
  FIRST_VALUE(A.channel) over(partition by A.userid order by B.ts rows between unbounded preceding and unbounded following) AS First_Channel,
  LAST_VALUE(A.channel) over(partition by A.userid order by B.ts rows between unbounded preceding and unbounded following) AS Last_Channel
FROM raw_data.user_session_channel A
LEFT JOIN raw_data.session_timestamp B ON A.sessionid = B.sessionid;""",
          'input_check': [],
          'output_check': []
}

Build_summary_v2.py

작성돼 있던 Build_summary_v2의 tables_load에 mau_summary, channel_summary를 추가해 주었다.

...
# this should be listed in dependency order (all in analytics)
tables_load = [
    'nps_summary',
    'mau_summary',
    'channel_summary'
]

dag_root_path = os.path.dirname(os.path.abspath(__file__))
redshift_summary.build_summary_table(dag_root_path, dag, tables_load, "redshift_dev_db")

결과

Build_Summary_v2