GCSHook GCP(Google Cloud Platform)을 사용하다가 Airflow에서 GCS(Google Cloud Storage)에 연동하여 데이터를 적재하는 방법에 대해서 찾아보다가 GCSHOOK을 이용하면 된다고 나와있어서 정리해본다. 마치 PostgresHook과 같이 Airlfow의 Connection이라는 기능에 접속정보를 입력하면 손쉽게 활용할 수 있을 뿐만 아니라 접속 정보를 코드안에 적지 않아도 된다는 장점이 있다. 우선 당연히 GCP에서 버킷이 생성되어 있어야한다. 또한 IAM 및 관리자 에서 권한 설정을 통해 원격으로 GCS에 접근할 수 있는 서비스 계정이라는 것을 생성해야된다. GCS 권한 설정 방법 서비스 계정 만들기를 시작하게 되면 해당 계정에 뷰어와 저장소 개체 관리자를..

Data Pipeline with Apache Airflow Apache Airflow 기반의 데이터 파이프라인 저서를 참고함 Airflow Sensor 소개 워크플로우를 트리거하는 여러가지 방법 중 외부 이벤트의 결과를 트리거로 워크플로우가 실행될 수 있게하는 방법에 대해서 소개하도록 한다. 예를 들면, 공유폴더에 파일이 업로드 되거나 Hive에 새로운 파티션이 생성됐을 때 워크플로우를 실행해야하는 경우가 있는데 이때 PythonSensor or FileSensor를 통해 새로운 데이터를 읽고 처리할 수 있다. 이러한 센서들은 특정 조건이 true인지 지속적으로 확인하고 true라면 성공하고 False면 상태가 true가 될 때까지 또는 타임아웃이 될 때까지 계속 확인한다. 여기서 센서의 타임아웃은 주..

Airflow + Postgresql 네이버 코스피 지수 크롤링 Airflow를 연습해보기 위해 네이버 증권에서 코스피 지수 일별 시세를 매일 수집해서 적재하는 배치를 만들어보려고 한다. 위의 그림과 같이 현재는 매일 수집하는 중이고 다른 방법으로 이러한 금융데이터를 수집 적재 해볼 생각이다. 1. URL 구하기 우선 크롤링을 진행할 네이버 증권의 KOSPI 일별 시세의 url을 보면 매일 새로운 데이터가 첫 페이지 맨 위에 장식된다. 그렇기 때문에 페이지와는 상관 없이 무조건 첫 페이지만 불러오면 된다. 우선 네이버 KOSPI 지수 일별 시세를 보여주는 페이지로 가보자. 네이버 증권의 코스피 페이지를 가보면 하단에 일별시세를 제목으로 하는 테이블이 있는데 이때 페이지를 넘기는 숫자에 커서를 올리면 UR..

Airflow에서는 동적 날짜 변수라는 Dag 파일의 실행과 관련된 메타 데이터를 제공한다. 이러한 날짜 변수들을 이용해 원하는 시점의 데이터를 가져온다던가 Backfill 기능을 활용하여 재적재를 용이하게 만드는 등 다양한 처리를 할 수 있다. 어떤 변수들이 있는지 Airflow 홈페이지에 자세히 나와있다.(날짜외의 다양한 메타 데이터를 제공한다!) 동적 날짜 변수 비교 Airflow는 ETL 툴로 개발되었다. 이때 동적으로 날짜라는 변수를 사용하는 것은 특정 날짜의 데이터를 추출한다거나 특정 날짜의 스케쥴에 대한 문제를 진단하는 등의 다양한 처리를 할 수 있다. 따라서 Airflow에서 자주 사용되는 것들을 실제로 실행시켜 보았다. import airflow from airflow import DAG..

Airflow를 설치하고 난 후 데이터를 적재하거나 읽어올 수 있는 RDB 툴이 필요했는데 로컬에 설치되어있는 Postgresql과 연동시키고 싶었다. 처음에는 host 파라미터에 localhost로 지정해주면 되지 않을까해서 Airflow Dag파일 작성하면서 localhost라고 지정해봤지만 실패하였다. 인터넷 검색을 통해 host.docker.internal 이라고 지정해주면 된다고 하지만 도커의 버전에 따라 작동안하는 경우도 있는 것 같고 나도 실패했다... 결국에는 docker 혹은 airflow 이미지 생성 시 설정파일들을 손보면 된다는데 처음에는 간단해보여서 로컬 Postgresql과 연동 시키려 했지만 너무 복잡했다. 그래서 빠르게 포기하고 새로운 postgresql container를 생..

회사에 입사 후 여러 사업에 참여하면서 데이터 분석을 할 때 굉장히 불편한 점을 많이 느꼈다. 분석과 데이터 엔지니어 일이 나눠져 있어서 불편한 경우가 한두가지가 아니었는데 새로운 데이터가 필요할 때면 항상 직접 처리하지 못하고 일일히 부탁해야된다는 점이 가장 불편했던거 같다. 현재는 NiFi라는 툴을 이용해 SI 사업을 진행 중이지만 앞으로 미래엔 점차 Airflow로 바뀌어가지 않을까하는 마음과 Python으로 데이터를 다룰 수 있다는 점에 Airflow를 배워보려고 한다. ("Apache Airflow 기반의 데이터 파이프라인" 이라는 책을 정리하면서 공부하려고 한다.) Airflow 장점 1. 파이썬 코드를 사용한 유연한 파이프라인 정의 Airflow는 파이썬 스크립트로 DAG(워크플로우 테스크)..