DataRami DataRami

아래 이미지는 현 회사에서 관리하고 있는 ETL 파이프라인 중 마케팅 Summary data에 관한 Airflow DAG 그래프입니다. 마케팅 데이터 파이프라인의 구성 1. 소스 데이터 입수 위의 DAG에는 포함되어있지 않지만, 광고 매체의 보고서 API를 호출하거나 서비스 DB에서 소스를 추출하는 스케줄도 에어플로우로 관리합니다. 2. 가공 및 적재 위의 DAG에서 가장 처음으로 실행되는 두 개의 Task인 userid_mapping >> userid_mapped_logs에 해당합니다. 이 Task는 log 데이터에서 로그인 전 생성되는 anonymous_id와 로그인 후 기록되는 user_id를 매핑해서 특정 유저가 로그인하기 전의 로그까지 해당 유저의 기록으로 연결해 주는 역할을 합니다. 다음으로 ..

airflow에 있는 SlackAPIPostOperator를 활용해서 slack에 메시지를 보내 보려고 합니다. 먼저 슬랙에서 app을 만들어 줍니다. 다음으로 incoming webhooks를 활성화합니다. 슬랙 채널에 앱을 연동한 뒤 airflow dag를 작성합니다. from datetime import datetime, timedelta import airflow from airflow import DAG from airflow.operators.slack_operator import SlackAPIPostOperator default_args = { 'owner': 'rami', 'depends_on_past': False, 'start_date': datetime(2020, 12, 17, 10..

python 파일로 이루어진 Airflow Dag를 관리하는 데에는 여러 방법이 있습니다. 가장 보편적으로 git을 이용하는 것, 그리고 데이터 분석 목적으로 사용하는 데는 이런 방법도 있다고 해서 적용해 봤습니다. 물론 git처럼 버전 관리에 유용한 건 아니지만 jupyter notebook으로 테스트를 하면서 동시에 dag를 띄울 수 있어서 복잡하지 않은 dag를 관리하는 데는 나쁘지 않은 방법이라는 생각이 듭니다. 일단, docker를 통해 설치해 둔 airflow 컨테이너에 접속합니다. docker exec -it airflow /bin/bash 앞의 airflow 세팅 과정에서 miniconda도 설치되어 있기 때문에 바로 jupyter lab 셋업으로 넘어갑니다. 2020/12/15 - [et..

Airflow Macro란? 지정된 변수를 호출해서 태스크에 적용할 수 있는 기능입니다. 만약 실행하는 코드에 날짜 설정이 now() 등으로 되어 있다면 backfill이나 과거의 작업을 재실행할 때 문제가 되는데, 이 때 {{execution_date}} 처럼 airflow task가 실행되는 날짜를 불러와서 적용할 수 있다는 것이 장점입니다. airflow.apache.org/docs/apache-airflow/stable/macros-ref.html Macros reference — Airflow Documentation airflow.apache.org BashOperator에서는 매크로를 그대로 기입하면 돼서 어려움이 없습니다. t2 = BashOperator( task_id='what_time..

2020/07/09 - [etc.] - Synology NAS -> Docker -> Jupyter Notebook 설치하기 Synology NAS -> Docker -> Jupyter Notebook 설치하기 설치환경 : Synology NAS DS218+ Synology NAS DSM -> Docker 패키지를 실행합니다. 레지스트리에서 ubuntu를 검색하고 적절한 이미지를 설치합니다. Jupyter Notebook 용도이기 때문에 설치 단계에서 8888포트를.. datarami.tistory.com 위 글과 같이 Synology NAS에 Docker 세팅을 하면서 컨테이너를 띄우기 전 접속 포트와 함께 로컬과 연결할 공유 폴더 볼륨 설정을 해 줍니다. 다음으로 아래 명령어를 차례로 입력해 airf..